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

Merge origin/master into worktree/agent-execution-context-rfc

Browse Source
This commit is contained in:
Yichen Jiang
2026-07-18 21:20:30 +08:00
parent a42c28cdc9 417248eed8
commit 0e4785cdb7
105 changed files with 3306 additions and 554 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -29,6 +29,7 @@ The config-driven `ctx.agentLoop.create()` path keeps its agent owned by the loo
```ts
interface Config {
maxParallelToolCalls?: number // default 10; 1 is serial
agents: Array<{
id: string // required
provider?: string
@@ -39,13 +40,13 @@ interface Config {
}
```
Configured agents start automatically. A model call requires both `provider` and `model`; `agent/request` may supply a missing pair before dispatch. `cwd` applies only to fresh sessions, while `resumeSessionId` retains persisted metadata. Configured agents use the deployment persona, and programmatic setup can shadow it per agent. This plugin supplies the per-agent `provider`, `model`, and `cwd` prompt variables; harness identity and deployment persona belong to `dsh-system-prompt`.
Configured agents start automatically. A model call requires both `provider` and `model`; `agent/request` may supply a missing pair before dispatch. `maxParallelToolCalls` bounds every agent's rolling pool for parallel-safe calls and defaults to `10`. `cwd` applies only to fresh sessions, while `resumeSessionId` retains persisted metadata. Configured agents use the deployment persona, and programmatic setup can shadow it per agent. This plugin supplies the per-agent `provider`, `model`, and `cwd` prompt variables; harness identity and deployment persona belong to `dsh-system-prompt`.
### Exported concrete class
- `ReactLoopAgent` — the concrete `Agent` implementation. Its inbox is a JavaScript native-private field, and one prepared session can be claimed by only one concrete driver. Everything observable happens through session events and the `agent/*` event taxonomy.
`Inbox`, `runLoop`, and the instance-bound publication/start controls are package-internal. The package root does not export them, and the package exports map exposes no `./src/*` escape hatch; lifecycle owners create agents through `ctx.agents` rather than constructing or starting the driver internals. `ReactLoopAgent.send()` and running `steer()` materialize content plus resolved source once as detached, deeply frozen lossless JSON, then share that accepted record between `agent/queued` and the inbox; malformed data throws before either boundary.
`Inbox`, `runLoop`, and the instance-bound publication/start controls are package-internal. The package root does not export them, and the package exports map exposes no `./src/*` escape hatch; lifecycle owners create agents through `ctx.agents` rather than constructing or starting the driver internals. `ReactLoopAgent.send()`, running `steer()`, and open-turn `inject()` materialize content plus resolved source once as detached, deeply frozen lossless JSON; malformed data throws before enqueue or append. An injection that arrives while the current step executes assistant tool calls stays in a FIFO until the batch settles; successful batches place it after the complete result batch, and interrupted batches drain it before the turn closes.
### Loop lifecycle (`loop.ts`)
@@ -57,6 +58,8 @@ Every provider call that reaches a successful finish appends exactly one `assist
Plugin failure ends the current turn, not the loop. Cancellation clears pending work and aborts the current step without leaking to the next prompt. 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.
### What belongs to plugins
Everything that goes beyond "call the model, run the tools, repeat" belongs to plugins listening on the event taxonomy:
@@ -83,7 +86,7 @@ Everything that goes beyond "call the model, run the tools, repeat" belongs to p
## Known Limitations and Deferred Work
- **Tool calls within a step execute sequentially** — parallel execution waits on concurrency-safety metadata in the tool contract (see `dsh-tools`).
- **Classification is unary** — calls whose safety depends on comparing siblings or resources must remain exclusive ([rationale](../../../docs/rfc/implemented/feature/2026-07-10-parallel-tool-call-execution.md)).
- **No resume-or-create policy on the config path** — config-driven `create()` starts a fresh `${id}-session-<uuid>` every run (`TODO(demo)`), and a config `resumeSessionId` whose resume fails logs a warning and creates no agent.
- **Config agents have no per-agent persona field or setup hook** — they use the deployment persona; scoped persona/tool composition is available only through the programmatic `ctx.agents.create()` / `resume()` factory options.
- **No built-in turn budget** — the default continuation is `continue` whenever a step had tool calls or steering; bounding a runaway turn requires an `agent/turn-continuation` force-stop plugin.

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

@@ -8,7 +8,7 @@
import type { Context } from 'cordis'
import { agentEvents } from '@deepseek-ai/dsh-agent'
import type { AgentId, AgentOptions, AgentStatus, InjectOptions, SendOptions } from '@deepseek-ai/dsh-agent'
import type { AgentId, AgentOptions, AgentStatus, HookContext, InjectOptions, SendOptions } from '@deepseek-ai/dsh-agent'
import type { Agent } from '@deepseek-ai/dsh-agent'
import { deepFreeze } from '@deepseek-ai/dsh-llm'
import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
@@ -54,15 +54,16 @@ export interface PreparedReactLoopAgent {
* @param id - the concrete agent identity.
* @param options - loop options for the agent.
* @param session - the prepared session the agent will own.
* @param maxParallelToolCalls - resolved in-flight cap for this agent.
* @returns the agent and closures bound only to that exact instance.
*/
export function prepareReactLoopAgent(
ctx: Context, id: AgentId, options: AgentOptions, session: Session,
ctx: Context, id: AgentId, options: AgentOptions, session: Session, maxParallelToolCalls: number,
): PreparedReactLoopAgent {
if (claimedDriverSessions.has(session)) {
throw new Error(`session "${session.id}" already has a concrete agent driver`)
}
const agent = new ReactLoopAgent(ctx, id, options, session)
const agent = new ReactLoopAgent(ctx, id, options, session, maxParallelToolCalls)
claimedDriverSessions.add(session)
const dispose = () => agent[stopDriver]()
return {
@@ -143,19 +144,27 @@ export class ReactLoopAgent implements Agent {
* the `disposed` transition fires and leave the promise hanging.
*/
private idleWaiters: (() => void)[] = []
/** Maximum parallel-safe calls allowed in one step. */
private readonly maxParallelToolCalls: number
/**
* Durability checkpoints started by idle {@link inject} calls. `inject()` is
* synchronous, so it cannot await them itself; the driver disposer drains
* this set before the lifecycle unregisters the agent or detaches its session.
*/
private pendingIdleFlushes = new Set<Promise<void>>()
/** Whether the current step is executing an assistant tool-call batch. */
private toolBatchActive = false
/** Open-turn injections waiting for the active assistant tool-call batch to close. */
private deferredInjections: HookContext[] = []
constructor(
private loopCtx: Context,
public readonly id: AgentId,
public readonly options: AgentOptions,
public readonly session: Session,
maxParallelToolCalls: number,
) {
this.maxParallelToolCalls = maxParallelToolCalls
const { promise, resolve } = Promise.withResolvers<void>()
this.disposed = promise
this.resolveDisposed = resolve
@@ -189,12 +198,11 @@ export class ReactLoopAgent implements Agent {
}
/**
* Accept one public send/steer payload as the exact detached record shared by
* the live notification and inbox. Lossless-JSON materialization reads every
* nested field once; deep freeze prevents an observer from rewriting queued
* work before the loop drains it.
* Accept one public message payload as a detached record. Lossless-JSON
* materialization reads every nested field once; deep freeze prevents later
* caller mutation before an inbox or deferred-injection queue drains it.
*/
private acceptInboxMessage(content: ContentBlock[], options?: SendOptions): InboxMessage {
private acceptMessage(content: ContentBlock[], options?: SendOptions): InboxMessage {
const source = this.resolveSource(options)
const accepted = snapshotJsonValue({ content, source })
if (accepted === undefined) {
@@ -203,6 +211,15 @@ export class ReactLoopAgent implements Agent {
return deepFreeze(accepted)
}
/** Detach one context before it can outlive its caller in the active-batch FIFO. */
private acceptContext(context: HookContext): HookContext {
const accepted = snapshotJsonValue(context)
if (accepted === undefined) {
throw new TypeError('agent context must be losslessly JSON-serializable')
}
return deepFreeze(accepted)
}
/** Reject a driving operation once teardown has synchronously closed the agent. */
private assertNotDisposed(): void {
if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`)
@@ -210,7 +227,7 @@ export class ReactLoopAgent implements Agent {
send(content: ContentBlock[], options?: SendOptions): void {
this.assertNotDisposed()
const accepted = this.acceptInboxMessage(content, options)
const accepted = this.acceptMessage(content, options)
this.#inbox.enqueue(accepted)
const info = { source: accepted.source, steering: false } as const
agentEvents(this.loopCtx, this).emit('agent/queued', accepted.content, info)
@@ -219,7 +236,7 @@ export class ReactLoopAgent implements Agent {
steer(content: ContentBlock[], options?: SendOptions): void {
this.assertNotDisposed()
if (this._status !== 'running') { this.send(content, options); return }
const accepted = this.acceptInboxMessage(content, options)
const accepted = this.acceptMessage(content, options)
this.#inbox.steer(accepted)
const info = { source: accepted.source, steering: true } as const
agentEvents(this.loopCtx, this).emit('agent/queued', accepted.content, info)
@@ -235,10 +252,15 @@ export class ReactLoopAgent implements Agent {
...options?.meta !== undefined ? { meta: options.meta } : {},
}
if (isTurnOpen(this.session)) {
// A turn is open in the LOG (decided from the log, not agent status —
// status can be `running` with no turn open): the context/message is
// turn-enclosed by that turn, so append it directly.
this.session.append('context/message', context, { surfaceOp: 'append' })
const accepted = this.acceptContext(context)
// Provider protocols require every assistant tool-call batch to be
// followed only by its tool results. Historical interrupted batches do
// not own new context; only the currently executing batch may defer it.
if (this.toolBatchActive) {
this.deferredInjections.push(accepted)
return
}
this.session.append('context/message', accepted, { surfaceOp: 'append' })
return
}
// No turn open: wrap the injection in a one-shot turn so every event stays
@@ -278,6 +300,34 @@ export class ReactLoopAgent implements Agent {
}
}
/** Append deferred open-turn injections after the loop closes a tool-result batch. */
private drainDeferredInjections(): void {
const pending = this.deferredInjections.splice(0)
for (const accepted of pending) {
this.session.append('context/message', accepted, { surfaceOp: 'append' })
}
}
/**
* Run one tool-call batch and drain its deferred context before settlement.
* The loop-owned acceptor remains valid after public disposal begins because
* the interrupted turn stays open until this batch settles.
*/
private async withToolBatch<T>(
run: (acceptContext: (context: HookContext) => void) => Promise<T>,
): Promise<T> {
this.toolBatchActive = true
const acceptContext = (context: HookContext): void => {
this.deferredInjections.push(this.acceptContext(context))
}
try {
return await run(acceptContext)
} finally {
this.toolBatchActive = false
this.drainDeferredInjections()
}
}
cancel(reason?: string): void {
// 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) {
@@ -335,6 +385,7 @@ export class ReactLoopAgent implements Agent {
this.driverStarted = true
this.done = this.loopCtx.agentExecution.run({ agent: this }, () => runLoop(this.loopCtx, this, {
inbox: this.#inbox,
maxParallelToolCalls: this.maxParallelToolCalls,
setStatus: (status) => { this.setStatus(status) },
setAbort: controller => void (this.currentAbort = controller),
disposed: this.disposed,
@@ -342,6 +393,7 @@ export class ReactLoopAgent implements Agent {
isCancelled: () => this.cancelRequested,
cancelReason: () => this.cancelReason,
clearCancel: () => { this.cancelRequested = false },
withToolBatch: run => this.withToolBatch(run),
// Pre-step cancellation re-parks without emitting a status transition.
settleIdle: () => { this.settleIdleWaiters() },
}))

6
packages/core/agent-loop/src/constants.ts Normal file
View File

@@ -0,0 +1,6 @@
/** Shared agent-loop scheduler defaults.
* @module dsh-agent-loop/constants
*/
/** Default maximum in-flight parallel-safe calls per agent step. */
export const DEFAULT_MAX_PARALLEL_TOOL_CALLS = 10

35
packages/core/agent-loop/src/index.ts
View File

@@ -33,6 +33,7 @@ import {
ReactLoopAgent,
} from './agent.ts'
import type { PreparedReactLoopAgent } from './agent.ts'
import { DEFAULT_MAX_PARALLEL_TOOL_CALLS } from './constants.ts'
export { ReactLoopAgent } from './agent.ts'
@@ -74,6 +75,15 @@ function signalAbortError(id: AgentId, signal: AbortSignal): Error {
return new Error(`agent "${id}" creation aborted`, { cause: signal.reason })
}
/** Resolve the deployment-wide scheduler cap at the owning config boundary. */
function resolveMaxParallelToolCalls(value: number | undefined): number {
const maxParallelToolCalls = value ?? DEFAULT_MAX_PARALLEL_TOOL_CALLS
if (!Number.isInteger(maxParallelToolCalls) || maxParallelToolCalls < 1) {
throw new Error('maxParallelToolCalls must be a positive integer')
}
return maxParallelToolCalls
}
/**
* Caller-owned create/resume transaction through rollback-covered publication
* and quiescent teardown. Resources remain private until the final registry
@@ -164,13 +174,13 @@ class AgentCreationTransaction {
}
/** Construct the driver and scope, then install their complete ordered lifecycle. */
prepare(options: AgentOptions, session: Session): ReactLoopAgent {
prepare(options: AgentOptions, session: Session, maxParallelToolCalls: number): ReactLoopAgent {
this.assertActive()
const gate = Promise.withResolvers<void>()
this.preparing = gate.promise
try {
this.session = session
const driver = prepareReactLoopAgent(this.loopCtx, this.id, options, session)
const driver = prepareReactLoopAgent(this.loopCtx, this.id, options, session, maxParallelToolCalls)
this.driver = driver
const agent = driver.agent
const scope = createScope(this.loopCtx, agent)
@@ -319,8 +329,15 @@ declare module 'cordis' {
}
}
/** Plugin configuration for declarative startup agents. */
export { DEFAULT_MAX_PARALLEL_TOOL_CALLS }
/** Agent-loop plugin configuration. */
export interface Config {
/**
* Maximum parallel-safe calls in flight per agent step. `1` is serial;
* omission defaults to {@link DEFAULT_MAX_PARALLEL_TOOL_CALLS}.
*/
maxParallelToolCalls?: number
/** Agents created or resumed at plugin startup. */
agents: (AgentOptions & {
/** Registry identity for the live agent. */
@@ -338,6 +355,7 @@ export class AgentLoop extends Service implements AgentFactory {
/** Runtime schema for declarative agents. */
static Config = z.object({
maxParallelToolCalls: z.number().step(1).min(1).default(DEFAULT_MAX_PARALLEL_TOOL_CALLS),
agents: z.array(z.object({
id: z.string().required(),
provider: z.string(),
@@ -348,11 +366,14 @@ export class AgentLoop extends Service implements AgentFactory {
}) as unknown as z<Config>
private readonly ownership: FactoryOwnership
/** Resolved concurrency cap for every driver created by this factory. */
private readonly maxParallelToolCalls: number
/** Plain holder prevents Cordis from re-tracing the factory's dependency context through a caller shadow. */
private readonly runtime: { ctx: Context }
constructor(ctx: Context, public config: Config) {
super(ctx, 'agentLoop')
this.maxParallelToolCalls = resolveMaxParallelToolCalls(config.maxParallelToolCalls)
this.ownership = new FactoryOwnership(ctx.fiber)
this.runtime = { ctx }
ctx.effect(() => () => this.ownership.dispose(), 'agentLoop.transactions()')
@@ -395,7 +416,7 @@ export class AgentLoop extends Service implements AgentFactory {
try {
const sessionId = SessionId(`${id}-session-${randomUUID()}`)
const session = loopCtx.sessions.prepare(sessionId, { meta })
const agent = transaction.prepare(options, session)
const agent = transaction.prepare(options, session, this.maxParallelToolCalls)
transaction.publish('startup')
return agent
} catch (error: unknown) {
@@ -413,6 +434,7 @@ export class AgentLoop extends Service implements AgentFactory {
* @returns the published handle.
*/
async createAgent(ownerCtx: Context, options: CreateAgentOptions): Promise<AgentHandle> {
const agentOptions = options.agentOptions ?? {}
const transaction = new AgentCreationTransaction(
this.runtime.ctx,
ownerCtx,
@@ -425,7 +447,7 @@ export class AgentLoop extends Service implements AgentFactory {
...options.seed === undefined ? {} : { seed: options.seed },
...options.meta === undefined ? {} : { meta: options.meta },
})
const agent = transaction.prepare(options.agentOptions ?? {}, session)
const agent = transaction.prepare(agentOptions, session, this.maxParallelToolCalls)
await transaction.waitFor(options.setup?.(agent.ctx))
transaction.assertActive()
return transaction.publish('startup')
@@ -457,6 +479,7 @@ export class AgentLoop extends Service implements AgentFactory {
persistence: SessionPersistence,
options: ResumeAgentOptions,
): Promise<AgentHandle> {
const agentOptions = options.agentOptions ?? {}
const transaction = new AgentCreationTransaction(
this.runtime.ctx,
ownerCtx,
@@ -476,7 +499,7 @@ export class AgentLoop extends Service implements AgentFactory {
...loaded.meta.seedLength === undefined ? {} : { seedLength: loaded.meta.seedLength },
},
})
const agent = transaction.prepare(options.agentOptions ?? {}, session)
const agent = transaction.prepare(agentOptions, session, this.maxParallelToolCalls)
await transaction.waitFor(options.setup?.(agent.ctx))
transaction.assertActive()
return transaction.publish('resume')

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

@@ -18,6 +18,7 @@ import type { TransmissionLog } from './request-log.ts'
import { renderPrompt } from '@deepseek-ai/dsh-system-prompt'
import type { PromptAssembly } from '@deepseek-ai/dsh-system-prompt'
import type {} from '@deepseek-ai/dsh-tools'
import { executeToolCalls } from './tool-calls.ts'
import type { ReactLoopAgent } from './agent.ts'
import type { Inbox } from './inbox.ts'
@@ -73,6 +74,8 @@ function stepFinishReason(finish: FinishReason): TurnEndReason | undefined {
export interface LoopHandle {
/** Native-private agent inbox handed to the driver only at internal startup. */
readonly inbox: Inbox
/** Maximum parallel-safe calls allowed in one step. */
readonly maxParallelToolCalls: number
setStatus(status: 'idle' | 'running'): void
setAbort(controller: AbortController | undefined): void
/** Resolves when the agent is disposed — unblocks the idle wait. */
@@ -86,6 +89,8 @@ export interface LoopHandle {
clearCancel(): void
/** Settle idle waiters when pre-running cancellation skips a turn, without emitting `agent/status`. */
settleIdle(): void
/** Run an active tool-call batch, accepting post-tool context into the FIFO drained before settlement. */
readonly withToolBatch: <T>(run: (acceptContext: (context: HookContext) => void) => Promise<T>) => Promise<T>
}
/**
@@ -331,7 +336,7 @@ async function runTurn(
let stepOutcome: { hadToolCalls: boolean; finish: FinishReason } | { error: Error }
try {
stepOutcome = await runStep(
ctx, events, agent, turn, step, assembly, fullSystemPrompt, boundaryMessages, transmission, abort.signal)
ctx, events, agent, handle, turn, step, assembly, fullSystemPrompt, boundaryMessages, transmission, abort.signal)
} catch (error: unknown) {
stepOutcome = { error: toError(error) }
} finally {
@@ -468,6 +473,7 @@ async function runStep(
ctx: Context,
events: AgentEventDispatch,
agent: ReactLoopAgent,
handle: LoopHandle,
turn: number,
step: number,
assembly: PromptAssembly,
@@ -556,58 +562,15 @@ async function runStep(
// empty chunk provenance for a contentless, usage-less provider response.
recordAssistantMessage(session, turn, step, header.config, assembledContent, message, assembler, chunkSeqs)
// Tool execution stays sequential; recheck abort around each normalized result.
// Dispatch may overlap; policy, durable results, and result context stay model-ordered.
const toolCalls = message.content.filter(block => block.type === 'tool-call')
// Buffer context until all results are appended to preserve call/result adjacency.
const pendingContext: HookContext[] = []
for (const call of toolCalls) {
/* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
const callEvent = session.append('tool/call', { turn, step, callId: call.id, name: call.name, arguments: call.arguments })
let parsedArguments: unknown
try {
parsedArguments = call.arguments ? JSON.parse(call.arguments) : {}
} catch {
parsedArguments = call.arguments
}
// TODO(pre-tool-input-rewrite): Keep logged history and live presentation aligned;
// see docs/rfc/proposed/feature/2026-06-30-pre-tool-input-rewrite.md.
const result = await ctx.tools.execute({
callId: call.id,
name: call.name,
arguments: parsedArguments,
agent,
signal,
})
session.append('tool/result', {
turn, step,
// Correlation comes from the immutable execution input; the result does
// not duplicate this authoritative transcript identity.
callId: call.id,
content: result.content,
isError: result.isError,
...result.error ? { error: result.error } : {},
// Persist tool-owned presentation data for replay.
...result.meta !== undefined ? { meta: result.meta } : {},
}, { surfaceOp: 'append', sourceEventSeqs: [callEvent.seq] })
pendingContext.push(...result.additionalContexts ?? [])
// The signal may flip while the tool is awaited.
/* v8 ignore start -- signal.reason default unreachable: cancel()/disposal always set it */
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
/* v8 ignore stop */
}
// Append buffered context after the complete result batch.
for (const context of pendingContext) {
agent.inject(context.content, {
source: context.source,
...context.envelope !== undefined ? { envelope: context.envelope } : {},
...context.meta !== undefined ? { meta: context.meta } : {},
})
}
return { hadToolCalls: toolCalls.length > 0, finish: assembler.finish }
if (toolCalls.length === 0) return { hadToolCalls: false, finish: assembler.finish }
return handle.withToolBatch(async (acceptContext) => {
await executeToolCalls(
ctx, agent, turn, step, toolCalls, signal, handle.maxParallelToolCalls, acceptContext,
)
return { hadToolCalls: true, finish: assembler.finish }
})
}
/** Preserve successful-call accounting without retaining output that result processing rejected. */

229
packages/core/agent-loop/src/tool-calls.ts Normal file
View File

@@ -0,0 +1,229 @@
/**
* Schedules one assistant step's tool calls. Exclusive calls form barriers;
* parallel calls use a bounded rolling pool and are reclassified before start.
* Dispatch may overlap, while policy, results, and result context remain
* model-ordered. Abort stops replenishment and drains started calls.
*
* Each started call records `tool/call`; `tool/result` commits in model order,
* preserving derived history when audit events interleave with earlier results.
* @module dsh-agent-loop/tool-calls
*/
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 type { ReactLoopAgent } from './agent.ts'
/** One tool call after argument parsing, ready to schedule. */
interface PlannedCall {
block: ToolCallBlock
exec: ToolExecutionInput
}
/** Settled dispatch awaiting model-order finalization. */
interface Slot {
exec: ToolRunContext
result: ToolExecutionResult
needsPost: boolean
}
/**
* Schedule one assistant step's tool calls by their live concurrency mode.
* Started calls receive ordered results. Abort drains them and rethrows after
* accepting their context into the batch FIFO owned by the caller.
*
* @param ctx - loop context that owns the tool registry.
* @param agent - agent and session receiving the call lifecycle.
* @param turn - current turn number.
* @param step - current step number.
* @param toolCalls - assistant calls in model order.
* @param signal - abort signal shared by the step.
* @param maxParallel - validated in-flight cap.
* @param acceptContext - accepts committed result context into the active batch.
*/
export async function executeToolCalls(
ctx: Context,
agent: ReactLoopAgent,
turn: number,
step: number,
toolCalls: ToolCallBlock[],
signal: AbortSignal,
maxParallel: number,
acceptContext: (context: HookContext) => void,
): Promise<void> {
const { session } = agent
// Inputs are distinct because tools/execute wrappers may replace `exec.signal`.
const planned: PlannedCall[] = toolCalls.map(block => ({
block,
exec: {
callId: block.id,
name: block.name,
arguments: parseArguments(block.arguments),
agent,
signal,
},
}))
let next = 0
while (next < planned.length) {
// Commit before classifying again so registry changes affect unstarted calls.
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- bounded by the loop condition
const first = planned[next]!
const mode = ctx.tools.executionMode(first.exec).kind
const group = mode === 'parallel' ? planned.slice(next) : [first]
next += await runGroup(ctx, session, turn, step, group, mode, signal, maxParallel, acceptContext)
}
}
/** Parse model arguments, preserving invalid JSON as text and mapping empty input to `{}`. */
function parseArguments(raw: string): unknown {
try {
return raw ? JSON.parse(raw) : {}
} catch {
return raw
}
}
/**
* Run one exclusive barrier or parallel pool. Later calls are reclassified
* before start; an exclusive reclassification waits for the current pool to
* drain and remains for the caller's next barrier. Results and contexts commit
* in model order. Abort stops starts, drains and commits started calls, accepts
* their contexts into the owning batch, and throws.
*/
async function runGroup(
ctx: Context,
session: Session,
turn: number,
step: number,
group: PlannedCall[],
mode: ToolExecutionMode['kind'],
signal: AbortSignal,
maxParallel: number,
acceptContext: (context: HookContext) => void,
): Promise<number> {
/* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
const slots: (Slot | undefined)[] = group.map(() => undefined)
// Started slots retain their tool/call seq for result provenance.
const callSeqs: number[] = group.map(() => -1)
let nextToStart = 0
let committed = 0
let started = 0
let aborted: boolean = signal.aborted
// `committed` advances only across contiguous model-order slots.
const commitReady = async (): Promise<void> => {
while (committed < group.length) {
const slot = slots[committed]
if (slot === undefined) break
const call = group[committed]
const result = slot.needsPost
? await ctx.tools[TOOL_REGISTRY_SCHEDULER].finalize(slot.exec, slot.result)
: ctx.tools[TOOL_REGISTRY_SCHEDULER].finish(slot.exec, slot.result)
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- bounded index
appendToolResult(session, turn, step, call!.block, result, callSeqs[committed]!)
for (const context of result.additionalContexts ?? []) acceptContext(context)
committed++
}
}
const inFlight = new Map<number, Promise<number>>()
const startCall = async (index: number): Promise<void> => {
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- bounded index
const call = group[index]!
callSeqs[index] = appendToolCall(session, turn, step, call.block)
started++
const prepared = await ctx.tools[TOOL_REGISTRY_SCHEDULER].prepare(call.exec)
switch (prepared.kind) {
case 'dispatch': {
const promise = ctx.tools[TOOL_REGISTRY_SCHEDULER].dispatch(prepared.exec).then((outcome) => {
slots[index] = { exec: prepared.exec, result: outcome.result, needsPost: outcome.kind === 'post-result' }
return index
})
inFlight.set(index, promise)
break
}
case 'post-result':
slots[index] = { exec: prepared.exec, result: prepared.result, needsPost: true }
break
case 'final-result':
slots[index] = { exec: prepared.exec, result: prepared.result, needsPost: false }
break
/* v8 ignore next -- closed-union exhaustiveness guard */
default:
assertNever(prepared, 'tool-call scheduler prepare result')
}
}
const fillPool = async (): Promise<void> => {
while (!aborted && nextToStart < group.length && inFlight.size < maxParallel) {
// Re-read later modes after ordered commits so registry changes can create a barrier.
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- bounded by the loop condition
const nextCall = group[nextToStart]!
if (nextToStart > 0 && mode === 'parallel'
&& ctx.tools.executionMode(nextCall.exec).kind !== 'parallel') break
await startCall(nextToStart)
nextToStart++
await commitReady()
// Abort may arrive while pre-execute awaits.
if (signal.aborted) aborted = true
}
}
// Ordered pre-execute may await; only dispatch/body overlaps.
// TODO: Drain every started call before rethrowing a scheduler error; tool
// bodies must not outlive the failed turn.
await fillPool()
while (inFlight.size > 0) {
const settledIndex = await Promise.race(inFlight.values())
inFlight.delete(settledIndex)
await commitReady()
// Abort may arrive while a tool or ordered commit awaits.
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
if (signal.aborted) aborted = true
await fillPool()
}
if (aborted) {
// Started calls and accepted context settle before the turn records the abort.
/* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
throw new Error(String(signal.reason ?? 'aborted'))
}
/* v8 ignore next -- unreachable: a non-aborted group commits every started call */
if (committed !== started) throw new Error('tool-call scheduler: uncommitted settled calls')
return started
}
/** Append a started call and return its provenance sequence. */
function appendToolCall(session: Session, turn: number, step: number, block: ToolCallBlock): number {
const event = session.append('tool/call', { turn, step, callId: block.id, name: block.name, arguments: block.arguments })
return event.seq
}
/** Append a model-ordered result linked to its call event. */
function appendToolResult(
session: Session,
turn: number,
step: number,
block: ToolCallBlock,
result: ToolExecutionResult,
callSeq: number,
): void {
session.append('tool/result', {
turn, step,
// Correlation stays with the loop's authoritative model-transcript call id;
// registry results deliberately do not duplicate it.
callId: block.id,
content: result.content,
isError: result.isError,
...result.error ? { error: result.error } : {},
// The tool's private presentation payload (e.g. a result-time diff),
// persisted so a UI bridge reproduces the card on replay.
...result.meta !== undefined ? { meta: result.meta } : {},
}, { surfaceOp: 'append', sourceEventSeqs: [callSeq] })
}

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

@@ -7,7 +7,7 @@ import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry from '@deepseek-ai/dsh-tools'
import AgentRegistry from '@deepseek-ai/dsh-agent'
import AgentExecutionProvider from '@deepseek-ai/dsh-agent-execution'
import AgentLoop, { ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
import AgentLoop, { DEFAULT_MAX_PARALLEL_TOOL_CALLS, ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
import { bindReactLoopAgentContext, prepareReactLoopAgent } from '../src/agent.ts'
import { MockAdapter, textResponse } from './mock-adapter.ts'
@@ -56,10 +56,14 @@ describe('ReactLoopAgent', () => {
await ctx.plugin(AgentExecutionProvider)
await ctx.plugin(SessionStore)
const session = ctx.sessions.create(SessionId('exclusive-driver'))
const prepared = prepareReactLoopAgent(ctx, AgentId('first-driver'), { provider: 'mock', model: 'mock' }, session)
const prepared = prepareReactLoopAgent(
ctx, AgentId('first-driver'), { provider: 'mock', model: 'mock' }, session, DEFAULT_MAX_PARALLEL_TOOL_CALLS,
)
expect(() => prepared.agent.ctx).toThrow('context is not bound')
expect(() => prepareReactLoopAgent(ctx, AgentId('second-driver'), { provider: 'mock', model: 'mock' }, session))
expect(() => prepareReactLoopAgent(
ctx, AgentId('second-driver'), { provider: 'mock', model: 'mock' }, session, DEFAULT_MAX_PARALLEL_TOOL_CALLS,
))
.toThrow('already has a concrete agent driver')
await prepared.dispose()
@@ -258,7 +262,9 @@ describe('ReactLoopAgent', () => {
await ctx.plugin(AgentExecutionProvider)
await ctx.plugin(SessionStore)
const session = ctx.sessions.create(SessionId('test'))
const prepared = prepareReactLoopAgent(ctx, AgentId('bare'), { provider: 'mock', model: 'mock' }, session)
const prepared = prepareReactLoopAgent(
ctx, AgentId('bare'), { provider: 'mock', model: 'mock' }, session, DEFAULT_MAX_PARALLEL_TOOL_CALLS,
)
const { agent } = prepared
prepared.markPublished()
@@ -276,7 +282,9 @@ describe('ReactLoopAgent', () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const session = ctx.sessions.create(SessionId('pre-start-dispose'))
const prepared = prepareReactLoopAgent(ctx, AgentId('pre-start-dispose'), { provider: 'mock', model: 'mock' }, session)
const prepared = prepareReactLoopAgent(
ctx, AgentId('pre-start-dispose'), { provider: 'mock', model: 'mock' }, session, DEFAULT_MAX_PARALLEL_TOOL_CALLS,
)
await prepared.dispose()
expect(prepared.agent.status).toBe('disposed')
@@ -374,7 +382,9 @@ describe('ReactLoopAgent', () => {
const adapter = new MockAdapter(['hang'])
ctx.llm.registerAdapter(['mock'], adapter)
const session = ctx.sessions.create(SessionId('bare'))
const prepared = prepareReactLoopAgent(ctx, AgentId('bare'), { provider: 'mock', model: 'mock' }, session)
const prepared = prepareReactLoopAgent(
ctx, AgentId('bare'), { provider: 'mock', model: 'mock' }, session, DEFAULT_MAX_PARALLEL_TOOL_CALLS,
)
const { agent } = prepared
prepared.markPublished()
const dispose = prepared.startDriver()

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

@@ -3,10 +3,10 @@ import { Context } from 'cordis'
import LlmService, { CallId, ContentBlock, MessageSource, 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 } from '@deepseek-ai/dsh-tools'
import ToolRegistry, { defineTool, type PostToolDecision } from '@deepseek-ai/dsh-tools'
import AgentRegistry, { AgentId, type ContinuationDecision } from '@deepseek-ai/dsh-agent'
import AgentExecutionProvider from '@deepseek-ai/dsh-agent-execution'
import AgentLoop, { ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
import AgentLoop, { DEFAULT_MAX_PARALLEL_TOOL_CALLS, ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
import { prepareReactLoopAgent } from '../src/agent.ts'
import * as Invariants from '@deepseek-ai/dsh-invariants'
import { maxTokensResponse, MockAdapter, textResponse, toolCallResponse } from './mock-adapter.ts'
@@ -251,6 +251,190 @@ describe('abort during tool execution ends the turn', () => {
expect(adapter.requests).toHaveLength(1) // no follow-up model call
expect(reasons).toEqual([{ kind: 'aborted', reason: 'user interrupt' }])
})
it('records context accepted before a tool-step abort in the same turn', async () => {
const adapter = new MockAdapter([toolCallResponse('c1', 'aborter', {})])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('a-abort-injection'), { provider: 'mock', model: 'mock' })
ctx.tools.register(defineTool({
name: 'aborter',
description: '',
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')
return [{ type: 'text', text: 'done' }]
},
}))
ctx.on('tools/post-execute', async (): Promise<PostToolDecision> => ({
kind: 'accept',
additionalContexts: [{
content: [{ type: 'text', text: 'accepted result context after abort' }],
source: { kind: 'plugin', plugin: 'test' },
}],
}))
send(agent, 'go')
await waitForIdle(ctx, agent)
const events = [...agent.session.events]
expect(events
.filter(event => event.type === 'tool/result' || event.type === 'context/message'
|| event.type === 'step/end' || event.type === 'turn/end')
.map(event => event.type))
.toEqual(['tool/result', 'context/message', 'context/message', 'step/end', 'turn/end'])
expect(events
.filter(event => event.type === 'context/message')
.map(event => event.data.content))
.toEqual([
[{ type: 'text', text: 'accepted before abort' }],
[{ type: 'text', text: 'accepted result context after abort' }],
])
})
it('records post-tool context when a later call aborts the batch', async () => {
const adapter = new MockAdapter([[
{ type: 'block-start', index: 0, blockType: 'tool-call' },
{ type: 'block-end', index: 0, block: { type: 'tool-call', id: CallId('c1'), name: 'first', arguments: '{}' } },
{ type: 'block-start', index: 1, blockType: 'tool-call' },
{ type: 'block-end', index: 1, block: { type: 'tool-call', id: CallId('c2'), name: 'aborter', arguments: '{}' } },
{ type: 'finish', reason: { kind: 'tool-calls' } },
] satisfies StreamChunk[]])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('a-later-abort-context'), { provider: 'mock', model: 'mock' })
ctx.tools.register(defineTool({
name: 'first',
description: '',
parameters: {},
async execute() {
return [{ type: 'text', text: 'first done' }]
},
}))
ctx.tools.register(defineTool({
name: 'aborter',
description: '',
parameters: {},
async execute() {
;(agent as unknown as { currentAbort?: AbortController }).currentAbort?.abort('user interrupt')
return [{ type: 'text', text: 'aborted' }]
},
}))
ctx.on('tools/post-execute', async (exec, _result, next): Promise<PostToolDecision> => {
if (exec.callId !== CallId('c1')) return next()
return {
kind: 'accept',
additionalContexts: [{
content: [{ type: 'text', text: 'accepted after first result' }],
source: { kind: 'plugin', plugin: 'test' },
}],
}
})
send(agent, 'go')
await waitForIdle(ctx, agent)
const events = [...agent.session.events]
expect(events
.filter(event => event.type === 'tool/result' || event.type === 'context/message'
|| event.type === 'step/end' || event.type === 'turn/end')
.map(event => event.type))
.toEqual(['tool/result', 'tool/result', 'context/message', 'step/end', 'turn/end'])
expect(events.find(event => event.type === 'context/message')?.data.content)
.toEqual([{ type: 'text', text: 'accepted after first result' }])
})
it('drains deferred context before disposal reaches quiescence', async () => {
const adapter = new MockAdapter([toolCallResponse('c1', 'waiter', {})])
const ctx = await harness(adapter)
const started = Promise.withResolvers<undefined>()
let agent!: ReactLoopAgent
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
agent = inner.agentLoop.create(AgentId('a-dispose-injection'), { provider: 'mock', model: 'mock' })
}, { inject: ['agentLoop'] }))
ctx.tools.register(defineTool({
name: 'waiter',
description: '',
parameters: {},
async execute(_args, exec) {
agent.inject([{ type: 'text', text: 'accepted before disposal' }], { source: { kind: 'plugin', plugin: 'test' } })
started.resolve(undefined)
const signal = exec.signal
if (!signal) throw new Error('tool execution signal is missing')
await new Promise<void>((resolve) => {
if (signal.aborted) resolve()
else signal.addEventListener('abort', () => { resolve() }, { once: true })
})
return [{ type: 'text', text: 'done' }]
},
}))
ctx.on('tools/post-execute', async (): Promise<PostToolDecision> => ({
kind: 'accept',
additionalContexts: [{
content: [{ type: 'text', text: 'accepted result context during disposal' }],
source: { kind: 'plugin', plugin: 'test' },
}],
}))
send(agent, 'go')
await started.promise
await fiber.dispose()
expect(agent.session.events
.filter(event => event.type === 'context/message')
.map(event => event.data.content))
.toEqual([
[{ type: 'text', text: 'accepted before disposal' }],
[{ type: 'text', text: 'accepted result context during disposal' }],
])
expect(agent.session.events.find(event => event.type === 'turn/end')?.data.reason)
.toEqual({ kind: 'disposed' })
})
it('limits injection deferral to the current tool batch', async () => {
const adapter = new MockAdapter([
[
{ type: 'block-start', index: 0, blockType: 'tool-call' },
{ type: 'block-end', index: 0, block: { type: 'tool-call', id: CallId('c1'), name: 'aborter', arguments: '{}' } },
{ type: 'block-start', index: 1, blockType: 'tool-call' },
{ type: 'block-end', index: 1, block: { type: 'tool-call', id: CallId('c2'), name: 'second', arguments: '{}' } },
{ type: 'finish', reason: { kind: 'tool-calls' } },
] satisfies StreamChunk[],
textResponse('later turn'),
])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('a-historical-tool-pair'), { provider: 'mock', model: 'mock' })
ctx.tools.register(defineTool({
name: 'aborter',
description: '',
parameters: {},
async execute() {
;(agent as unknown as { currentAbort?: AbortController }).currentAbort?.abort('user interrupt')
return [{ type: 'text', text: 'done' }]
},
}))
ctx.tools.register(defineTool({
name: 'second',
description: '',
parameters: {},
async execute() {
return [{ type: 'text', text: 'must not run' }]
},
}))
send(agent, 'leave an unmatched historical call')
await waitForIdle(ctx, agent)
ctx.on('agent/pre-step', (subject, turn) => {
if (subject === agent && turn === 2) {
agent.inject([{ type: 'text', text: 'new turn context' }], { source: { kind: 'plugin', plugin: 'test' } })
}
})
send(agent, 'start a text-only turn')
await waitForIdle(ctx, agent)
expect(agent.session.events.find(event => event.type === 'context/message')?.data.content)
.toEqual([{ type: 'text', text: 'new turn context' }])
expect(JSON.stringify(adapter.requests[1]?.messages)).toContain('new turn context')
})
})
describe('steering from late extension points is never stranded', () => {
@@ -641,7 +825,9 @@ describe('turn numbering continues across seeded sessions', () => {
ctx2.llm.registerAdapter(['mock'], second)
const seeded = ctx2.sessions.create(SessionId('forked'), { seed: [...agent.session.events] })
const prepared = prepareReactLoopAgent(ctx2, AgentId('forked-agent'), { provider: 'mock', model: 'mock' }, seeded)
const prepared = prepareReactLoopAgent(
ctx2, AgentId('forked-agent'), { provider: 'mock', model: 'mock' }, seeded, DEFAULT_MAX_PARALLEL_TOOL_CALLS,
)
const forked = prepared.agent
prepared.markPublished()
ctx2.effect(() => prepared.startDriver())

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

@@ -412,22 +412,30 @@ describe('agent loop', () => {
expect(requestText).not.toContain('<context source=')
})
it('inject() while running appends into the open turn (no extra synthetic turn)', async () => {
it('defers inject() during tool execution until after the tool result', async () => {
const adapter = new MockAdapter([
toolCallResponse('c1', 'noticer', {}, 'calling'),
textResponse('done'),
])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('a1'), { provider: 'mock', model: 'mock' })
// A tool that injects mid-execution: at this point the agent is running, so
// inject must append the context/message into the ALREADY-open turn rather
// than wrap it in its own one-shot turn.
let visibleDuringTool = false
const meta = { kind: 'deferred-test', version: 1 }
ctx.tools.register(defineTool({
name: 'noticer',
description: 'injects a notice',
parameters: {},
async execute() {
agent.inject([{ type: 'text', text: 'mid-turn notice' }], { source: { kind: 'plugin', plugin: 'x' } })
await Promise.resolve()
const first = { type: 'text' as const, text: 'mid-turn notice' }
agent.inject([first], {
source: { kind: 'plugin', plugin: 'x' },
envelope: 'raw',
meta,
})
first.text = 'mutated after inject'
agent.inject([{ type: 'text', text: 'second notice' }], { source: { kind: 'plugin', plugin: 'x' } })
visibleDuringTool = agent.session.events.some(e => e.type === 'context/message')
return [{ type: 'text', text: 'ok' }]
},
}))
@@ -435,13 +443,67 @@ describe('agent loop', () => {
send(agent, 'go')
await waitForIdle(ctx, agent)
// Exactly ONE turn ran (no synthetic injection turn), and the mid-turn
// context/message sits inside it.
expect(visibleDuringTool).toBe(false)
// The injection stays in the open turn, but its user-role context cannot
// split the assistant tool call from the provider's tool-result message.
const turnStarts = agent.session.events.filter(e => e.type === 'turn/start')
expect(turnStarts).toHaveLength(1)
const ts0 = turnStarts[0]!
expect(ts0.type === 'turn/start' && ts0.data.trigger.kind).toBe('message')
expect(agent.session.events.some(e => e.type === 'context/message')).toBe(true)
const result = agent.session.events.find(e => e.type === 'tool/result')!
const contexts = agent.session.events.filter(e => e.type === 'context/message')
expect(contexts).toHaveLength(2)
expect(result.seq).toBeLessThan(contexts[0]!.seq)
expect(contexts[0]?.type === 'context/message' && contexts[0].data).toMatchObject({
envelope: 'raw',
meta,
})
expect(contexts.flatMap(event => event.type === 'context/message' ? event.data.content : []))
.toEqual([
{ type: 'text', text: 'mid-turn notice' },
{ type: 'text', text: 'second notice' },
])
const secondRequest = adapter.requests[1]!.messages
const resultIndex = secondRequest.findIndex(message =>
message.content.some(block => block.type === 'tool-result'))
const contextIndexes = secondRequest.flatMap((message, index) =>
message.content.some(block => block.type === 'text'
&& (block.text.includes('mid-turn notice') || block.text.includes('second notice')))
? [index]
: [])
expect(resultIndex).toBeGreaterThanOrEqual(0)
expect(contextIndexes).toHaveLength(2)
expect(contextIndexes.every(index => index > resultIndex)).toBe(true)
})
it('rejects non-JSON context before it enters the active tool-batch FIFO', async () => {
const adapter = new MockAdapter([
toolCallResponse('c1', 'invalid-injector', {}, 'calling'),
textResponse('done'),
])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('invalid-context'), { provider: 'mock', model: 'mock' })
ctx.tools.register(defineTool({
name: 'invalid-injector',
description: 'attempts an invalid context injection',
parameters: {},
async execute() {
expect(() => {
agent.inject([{ type: 'text', text: 'invalid' }], {
source: { kind: 'plugin', plugin: 'test' },
meta: { bigint: 1n } as never,
})
}).toThrow('agent context must be losslessly JSON-serializable')
return [{ type: 'text', text: 'rejected invalid context' }]
},
}))
send(agent, 'go')
await waitForIdle(ctx, agent)
expect(agent.session.events.some(event => event.type === 'context/message')).toBe(false)
})
it('agent/turn-continuation can force-continue (/loop pattern) and force-stop', async () => {

574
packages/core/agent-loop/tests/tool-calls.spec.ts Normal file
View File

@@ -0,0 +1,574 @@
/**
* Exercises scheduler ordering and cancellation with deterministic gated tools.
* ACP goldens own transcript-facing coverage.
*/
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import { CallId, StreamChunk } from '@deepseek-ai/dsh-llm'
import SessionStore, { SessionEvent } 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 AgentRegistry, { AgentId } from '@deepseek-ai/dsh-agent'
import AgentExecutionProvider from '@deepseek-ai/dsh-agent-execution'
import AgentLoop, { ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
import { MockAdapter, textResponse } from './mock-adapter.ts'
async function harness(adapter: MockAdapter, maxParallelToolCalls?: number) {
const ctx = new Context()
await ctx.plugin(LlmService)
await ctx.plugin(SessionStore)
await ctx.plugin(SystemPrompt, { persona: '' })
await ctx.plugin(ToolRegistry)
await ctx.plugin(AgentRegistry)
await ctx.plugin(AgentExecutionProvider)
await ctx.plugin(AgentLoop, {
agents: [],
...maxParallelToolCalls === undefined ? {} : { maxParallelToolCalls },
})
ctx.llm.registerAdapter(['mock'], adapter)
return ctx
}
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() }
})
})
}
function events(agent: ReactLoopAgent): SessionEvent[] {
return [...agent.session.events]
}
/** Build one assistant response containing the supplied tool calls. */
function multiCall(calls: { id: string; name: string; args: object }[]): StreamChunk[] {
const chunks: StreamChunk[] = []
calls.forEach((call, index) => {
chunks.push(
{ type: 'block-start', index, blockType: 'tool-call' },
{ type: 'block-end', index, block: { type: 'tool-call', id: CallId(call.id), name: call.name, arguments: JSON.stringify(call.args) } },
)
})
chunks.push(
{ type: 'usage', usage: { inputTokens: 5, outputTokens: 5 } },
{ type: 'finish', reason: { kind: 'tool-calls' } },
)
return chunks
}
/** A tool whose calls block until the test releases them by callId. */
function gatedTool(name: string, parallel: boolean) {
const gates = new Map<string, () => void>()
const started: string[] = []
const tool = defineTool({
name,
description: `gated ${name}`,
parameters: { id: { type: 'string', required: true } },
...parallel ? { isConcurrencySafe: () => true } : {},
async execute(args) {
started.push(args.id)
await new Promise<void>((resolve) => { gates.set(args.id, resolve) })
return [{ type: 'text', text: `done-${args.id}` }]
},
})
return {
tool,
started,
release(id: string) { gates.get(id)?.(); gates.delete(id) },
pending() { return [...gates.keys()] },
}
}
function gatedParallelTool(name: string) {
return gatedTool(name, true)
}
function gatedExclusiveTool(name: string) {
return gatedTool(name, false)
}
/** Poll until `predicate` holds, letting microtasks/timers drain between checks. */
async function until(predicate: () => boolean): Promise<void> {
for (let i = 0; i < 1000 && !predicate(); i++) await new Promise(r => setTimeout(r, 0))
if (!predicate()) throw new Error('until: condition never held')
}
describe('tool-call scheduler: grouping and barriers', () => {
it('runs parallel-safe siblings concurrently (all start before any completes)', async () => {
const adapter = new MockAdapter([
multiCall([{ id: 'c1', name: 'p', args: { id: '1' } }, { id: 'c2', name: 'p', args: { id: '2' } }, { id: 'c3', name: 'p', args: { id: '3' } }]),
textResponse('done'),
])
const ctx = await harness(adapter)
const gated = gatedParallelTool('p')
ctx.tools.register(gated.tool)
const agent = ctx.agentLoop.create(AgentId('a1'), { provider: 'mock', model: 'mock' })
agent.send([{ type: 'text', text: 'go' }])
await until(() => gated.started.length === 3)
expect(gated.started).toEqual(['1', '2', '3'])
gated.release('1'); gated.release('2'); gated.release('3')
await waitForIdle(ctx, agent)
})
it('an exclusive call between two parallel-safe calls forms a barrier (3 groups)', async () => {
const order: string[] = []
const adapter = new MockAdapter([
multiCall([
{ id: 'c1', name: 'r', args: { id: 'A1' } },
{ id: 'c2', name: 'w', args: { id: 'A2' } },
{ id: 'c3', name: 'r', args: { id: 'A3' } },
]),
textResponse('done'),
])
const ctx = await harness(adapter)
ctx.tools.register(defineTool({
name: 'r', description: 'read', parameters: { id: { type: 'string', required: true } },
isConcurrencySafe: () => true,
async execute(args) { order.push(`r-start-${args.id}`); order.push(`r-end-${args.id}`); return [{ type: 'text', text: 'r' }] },
}))
ctx.tools.register(defineTool({
name: 'w', description: 'write', parameters: { id: { type: 'string', required: true } },
async execute(args) { order.push(`w-${args.id}`); return [{ type: 'text', text: 'w' }] },
}))
const agent = ctx.agentLoop.create(AgentId('a1'), { provider: 'mock', model: 'mock' })
agent.send([{ type: 'text', text: 'go' }])
await waitForIdle(ctx, agent)
expect(order).toEqual(['r-start-A1', 'r-end-A1', 'w-A2', 'r-start-A3', 'r-end-A3'])
})
it('reclassifies pending calls after an exclusive barrier replaces their tool', async () => {
const adapter = new MockAdapter([
multiCall([
{ id: 'c1', name: 'replace', args: { id: '0' } },
{ id: 'c2', name: 'x', args: { id: '1' } },
{ id: 'c3', name: 'x', args: { id: '2' } },
]),
textResponse('done'),
])
const ctx = await harness(adapter)
const replacement = gatedExclusiveTool('x')
const disposeSafe = ctx.tools.register(defineTool({
name: 'x',
description: 'initially safe',
parameters: { id: { type: 'string', required: true } },
isConcurrencySafe: () => true,
async execute(args) { return [{ type: 'text', text: `old-${args.id}` }] },
}))
ctx.tools.register(defineTool({
name: 'replace',
description: 'replace x',
parameters: { id: { type: 'string', required: true } },
async execute() {
disposeSafe()
ctx.tools.register(replacement.tool)
return [{ type: 'text', text: 'replaced' }]
},
}))
const agent = ctx.agentLoop.create(AgentId('a1'), { provider: 'mock', model: 'mock' })
agent.send([{ type: 'text', text: 'go' }])
await until(() => replacement.started.length === 1)
await new Promise(r => setTimeout(r, 5))
expect(replacement.started).toEqual(['1'])
replacement.release('1')
await until(() => replacement.started.length === 2)
expect(replacement.started).toEqual(['1', '2'])
replacement.release('2')
await waitForIdle(ctx, agent)
})
it('stops replenishing when a result observer makes the next call exclusive', async () => {
const adapter = new MockAdapter([
multiCall([
{ id: 'c1', name: 'x', args: { id: '1' } },
{ id: 'c2', name: 'x', args: { id: '2' } },
{ id: 'c3', name: 'x', args: { id: '3' } },
]),
textResponse('done'),
])
const ctx = await harness(adapter, 2)
const initial = gatedParallelTool('x')
const replacement = gatedExclusiveTool('x')
const disposeInitial = ctx.tools.register(initial.tool)
ctx.on('tools/result', (exec) => {
if (exec.callId !== CallId('c1')) return
disposeInitial()
ctx.tools.register(replacement.tool)
})
const agent = ctx.agentLoop.create(AgentId('a1'), { provider: 'mock', model: 'mock' })
agent.send([{ type: 'text', text: 'go' }])
await until(() => initial.started.length === 2)
initial.release('1')
await until(() => events(agent).some(event =>
event.type === 'tool/result' && event.data.callId === CallId('c1')))
await new Promise(r => setTimeout(r, 5))
expect(replacement.started).toEqual([])
initial.release('2')
await until(() => replacement.started.length === 1)
expect(replacement.started).toEqual(['3'])
replacement.release('3')
await waitForIdle(ctx, agent)
})
})
describe('tool-call scheduler: model-order results despite out-of-order settlement', () => {
it('commits tool/result in model order even when a later call settles first', async () => {
const adapter = new MockAdapter([
multiCall([{ id: 'c1', name: 'p', args: { id: '1' } }, { id: 'c2', name: 'p', args: { id: '2' } }]),
textResponse('done'),
])
const ctx = await harness(adapter)
const gated = gatedParallelTool('p')
ctx.tools.register(gated.tool)
const agent = ctx.agentLoop.create(AgentId('a1'), { provider: 'mock', model: 'mock' })
agent.send([{ type: 'text', text: 'go' }])
await until(() => gated.started.length === 2)
gated.release('2')
await new Promise(r => setTimeout(r, 5))
const beforeFirst = events(agent).filter(e => e.type === 'tool/result')
expect(beforeFirst).toEqual([])
gated.release('1')
await waitForIdle(ctx, agent)
const results = events(agent).filter(e => e.type === 'tool/result')
expect(results.map(e => e.data.callId)).toEqual([CallId('c1'), CallId('c2')])
})
it('derived history pairs calls in model order regardless of tool/call log interleaving', async () => {
const adapter = new MockAdapter([
multiCall([{ id: 'c1', name: 'p', args: { id: '1' } }, { id: 'c2', name: 'p', args: { id: '2' } }]),
textResponse('done'),
])
const ctx = await harness(adapter)
const gated = gatedParallelTool('p')
ctx.tools.register(gated.tool)
const agent = ctx.agentLoop.create(AgentId('a1'), { provider: 'mock', model: 'mock' })
agent.send([{ type: 'text', text: 'go' }])
await until(() => gated.started.length === 2)
gated.release('2'); gated.release('1')
await waitForIdle(ctx, agent)
const messages = agent.session.deriveMessages()
const toolResults = messages.flatMap(m => m.content.filter(b => b.type === 'tool-result'))
expect(toolResults.map(b => b.toolCallId)).toEqual([CallId('c1'), CallId('c2')])
})
})
describe('tool-call scheduler: rolling pool honors maxParallelToolCalls', () => {
it('rejects invalid global maxParallelToolCalls config at plugin load', async () => {
await expect(harness(new MockAdapter([]), 0)).rejects.toThrow()
await expect(harness(new MockAdapter([]), 1.5)).rejects.toThrow()
})
it('defensively rejects invalid caps when direct construction bypasses the config schema', () => {
expect(() => new AgentLoop(new Context(), { agents: [], maxParallelToolCalls: 0 }))
.toThrow('maxParallelToolCalls must be a positive integer')
expect(() => new AgentLoop(new Context(), { agents: [], maxParallelToolCalls: 1.5 }))
.toThrow('maxParallelToolCalls must be a positive integer')
})
it('defaults the cap when direct construction bypasses the config schema', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
await ctx.plugin(SessionStore)
await ctx.plugin(SystemPrompt, { persona: '' })
await ctx.plugin(ToolRegistry)
await ctx.plugin(AgentRegistry)
expect(() => new AgentLoop(ctx, { agents: [] })).not.toThrow()
await ctx.fiber.dispose()
})
it('starts at most the cap, replenishing as calls settle', async () => {
const adapter = new MockAdapter([
multiCall([1, 2, 3, 4].map(n => ({ id: `c${n}`, name: 'p', args: { id: String(n) } }))),
textResponse('done'),
])
const ctx = await harness(adapter, 2)
const gated = gatedParallelTool('p')
ctx.tools.register(gated.tool)
const agent = ctx.agentLoop.create(AgentId('a1'), { provider: 'mock', model: 'mock' })
agent.send([{ type: 'text', text: 'go' }])
await until(() => gated.started.length === 2)
await new Promise(r => setTimeout(r, 5))
expect(gated.started).toEqual(['1', '2'])
gated.release('1')
await until(() => gated.started.length === 3)
expect(gated.started).toEqual(['1', '2', '3'])
expect(events(agent)
.filter(e => e.type === 'tool/call' || e.type === 'tool/result')
.map(e => `${e.type}:${String(e.data.callId)}`)
.slice(0, 4))
.toEqual(['tool/call:c1', 'tool/call:c2', 'tool/result:c1', 'tool/call:c3'])
gated.release('2'); gated.release('3')
await until(() => gated.started.length === 4)
gated.release('4')
await waitForIdle(ctx, agent)
expect(events(agent).filter(e => e.type === 'tool/result').map(e => e.data.callId))
.toEqual([CallId('c1'), CallId('c2'), CallId('c3'), CallId('c4')])
})
it('maxParallelToolCalls: 1 is fully serial (no second start before the first settles)', async () => {
const adapter = new MockAdapter([
multiCall([{ id: 'c1', name: 'p', args: { id: '1' } }, { id: 'c2', name: 'p', args: { id: '2' } }]),
textResponse('done'),
])
const ctx = await harness(adapter, 1)
const gated = gatedParallelTool('p')
ctx.tools.register(gated.tool)
const agent = ctx.agentLoop.create(AgentId('a1'), { provider: 'mock', model: 'mock' })
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 until(() => gated.started.length === 2)
gated.release('2')
await waitForIdle(ctx, agent)
})
it('applies the configured cap to every factory-created agent', async () => {
const adapter = new MockAdapter([
multiCall([{ id: 'c1', name: 'p', args: { id: '1' } }, { id: 'c2', name: 'p', args: { id: '2' } }]),
textResponse('done'),
])
const ctx = new Context()
await ctx.plugin(LlmService)
await ctx.plugin(SessionStore)
await ctx.plugin(SystemPrompt, { persona: '' })
await ctx.plugin(ToolRegistry)
await ctx.plugin(AgentRegistry)
await ctx.plugin(AgentExecutionProvider)
await ctx.plugin(AgentLoop, { agents: [], maxParallelToolCalls: 1 })
ctx.llm.registerAdapter(['mock'], adapter)
const gated = gatedParallelTool('p')
ctx.tools.register(gated.tool)
const agent = ctx.agentLoop.create(AgentId('a1'), { provider: 'mock', model: 'mock' })
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 until(() => gated.started.length === 2)
gated.release('2')
await waitForIdle(ctx, agent)
})
})
describe('tool-call scheduler: ordered middleware and additional contexts', () => {
it('tools/pre-execute and tools/post-execute observe model call order', async () => {
const adapter = new MockAdapter([
multiCall([{ id: 'c1', name: 'p', args: { id: '1' } }, { id: 'c2', name: 'p', args: { id: '2' } }, { id: 'c3', name: 'p', args: { id: '3' } }]),
textResponse('done'),
])
const ctx = await harness(adapter)
const gated = gatedParallelTool('p')
ctx.tools.register(gated.tool)
const pre: string[] = []
const post: string[] = []
ctx.on('tools/pre-execute', async (exec, next): Promise<PreToolDecision> => { pre.push(String(exec.callId)); return next() })
ctx.on('tools/post-execute', async (exec, _result, next): Promise<PostToolDecision> => { post.push(String(exec.callId)); return next() })
const agent = ctx.agentLoop.create(AgentId('a1'), { provider: 'mock', model: 'mock' })
agent.send([{ type: 'text', text: 'go' }])
await until(() => gated.started.length === 3)
gated.release('3'); gated.release('2'); gated.release('1')
await waitForIdle(ctx, agent)
expect(pre).toEqual([CallId('c1'), CallId('c2'), CallId('c3')].map(String))
expect(post).toEqual([CallId('c1'), CallId('c2'), CallId('c3')].map(String))
})
it('injects additional contexts in model call order, not settlement order', async () => {
const adapter = new MockAdapter([
multiCall([{ id: 'c1', name: 'p', args: { id: '1' } }, { id: 'c2', name: 'p', args: { id: '2' } }]),
textResponse('done'),
])
const ctx = await harness(adapter, 2)
const gated = gatedParallelTool('p')
ctx.tools.register(gated.tool)
ctx.on('tools/post-execute', async (exec, _result): Promise<PostToolDecision> =>
({ kind: 'accept', additionalContexts: [{ content: [{ type: 'text', text: `ctx-${exec.callId}` }], source: { kind: 'plugin', plugin: 'p' } }] }))
const agent = ctx.agentLoop.create(AgentId('a1'), { provider: 'mock', model: 'mock' })
agent.send([{ type: 'text', text: 'go' }])
await until(() => gated.started.length === 2)
gated.release('2'); gated.release('1')
await waitForIdle(ctx, agent)
const log = events(agent)
const contextTexts = log.filter(e => e.type === 'context/message')
.map(e => (e.data.content[0] as { text: string }).text)
expect(contextTexts).toEqual(['ctx-c1', 'ctx-c2'])
const lastResult = log.findLastIndex(e => e.type === 'tool/result')
const firstContext = log.findIndex(e => e.type === 'context/message')
expect(lastResult).toBeLessThan(firstContext)
})
it('orders pre-execute denials and errors without dispatching them', async () => {
const adapter = new MockAdapter([
multiCall([
{ id: 'c1', name: 'p', args: { id: '1' } },
{ id: 'c2', name: 'p', args: { id: '2' } },
{ id: 'c3', name: 'p', args: { id: '3' } },
]),
textResponse('done'),
])
const ctx = await harness(adapter)
const gated = gatedParallelTool('p')
ctx.tools.register(gated.tool)
const post: string[] = []
ctx.on('tools/pre-execute', async (exec, next): Promise<PreToolDecision> => {
if (exec.callId === CallId('c2')) return { kind: 'deny', reason: 'blocked by policy' }
if (exec.callId === CallId('c3')) throw new Error('pre exploded')
return next()
})
ctx.on('tools/post-execute', async (exec, _result, next): Promise<PostToolDecision> => {
post.push(String(exec.callId))
return next()
})
const agent = ctx.agentLoop.create(AgentId('a1'), { provider: 'mock', model: 'mock' })
agent.send([{ type: 'text', text: 'go' }])
await until(() => gated.started.length === 1)
gated.release('1')
await waitForIdle(ctx, agent)
expect(gated.started).toEqual(['1'])
expect(post).toEqual(['c1', 'c2'])
const results = events(agent).filter(e => e.type === 'tool/result')
expect(results.map(e => e.data.callId)).toEqual([CallId('c1'), CallId('c2'), CallId('c3')])
expect((results[1]!.data.content[0] as { text: string }).text).toContain('blocked by policy')
expect((results[2]!.data.content[0] as { text: string }).text).toContain('pre exploded')
})
})
describe('tool-call scheduler: abort handling', () => {
it('starts no calls when the signal is already aborted before a parallel group', 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'),
])
const ctx = await harness(adapter)
const gated = gatedParallelTool('p')
ctx.tools.register(gated.tool)
const agent = ctx.agentLoop.create(AgentId('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.send([{ type: 'text', text: 'go' }])
await waitForIdle(ctx, agent)
expect(gated.started).toEqual([])
expect(events(agent).filter(e => e.type === 'tool/call')).toEqual([])
expect(events(agent).filter(e => e.type === 'tool/result')).toEqual([])
})
it('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'),
])
const ctx = await harness(adapter)
const gated = gatedParallelTool('p')
ctx.tools.register(gated.tool)
const agent = ctx.agentLoop.create(AgentId('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')
}
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(events(agent).filter(e => e.type === 'tool/call').map(e => e.data.callId))
.toEqual([CallId('c1')])
expect(events(agent).filter(e => e.type === 'tool/result').map(e => e.data.callId))
.toEqual([CallId('c1')])
})
it('stops replenishing after abort, commits started results, and drains accepted additional contexts', async () => {
const adapter = new MockAdapter([
multiCall([1, 2, 3, 4].map(n => ({ id: `c${n}`, name: 'p', args: { id: String(n) } }))),
textResponse('should never be requested'),
])
const ctx = await harness(adapter, 2)
const gated = gatedParallelTool('p')
ctx.tools.register(gated.tool)
ctx.on('tools/post-execute', async (exec, _result, next): Promise<PostToolDecision> => ({
...await next(),
additionalContexts: [{ content: [{ type: 'text', text: `ctx-${exec.callId}` }], source: { kind: 'plugin', plugin: 'p' } }],
}))
const agent = ctx.agentLoop.create(AgentId('a1'), { provider: 'mock', model: 'mock' })
agent.send([{ type: 'text', text: 'go' }])
await until(() => gated.started.length === 2)
;(agent as unknown as { currentAbort?: AbortController }).currentAbort?.abort('stop now')
gated.release('1')
gated.release('2')
await waitForIdle(ctx, agent)
expect(gated.started).toEqual(['1', '2'])
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')])
const settled = events(agent).filter(e => e.type === 'tool/result' || e.type === 'context/message')
expect(settled.map(e => e.type))
.toEqual(['tool/result', 'tool/result', 'context/message', 'context/message'])
expect(settled.filter(e => e.type === 'context/message')
.map(e => (e.data.content[0] as { text: string }).text))
.toEqual(['ctx-c1', 'ctx-c2'])
})
it('does not run an exclusive barrier after a parallel group aborts', async () => {
const adapter = new MockAdapter([
multiCall([
{ id: 'c1', name: 'p', args: { id: '1' } },
{ id: 'c2', name: 'p', args: { id: '2' } },
{ id: 'c3', name: 'x', args: { id: '3' } },
]),
textResponse('should never be requested'),
])
const ctx = await harness(adapter, 2)
const gated = gatedParallelTool('p')
const exclusive: string[] = []
ctx.tools.register(gated.tool)
ctx.tools.register(defineTool({
name: 'x',
description: 'exclusive',
parameters: { id: { type: 'string', required: true } },
async execute(args) { exclusive.push(args.id); return [{ type: 'text', text: 'x' }] },
}))
const agent = ctx.agentLoop.create(AgentId('a1'), { provider: 'mock', model: 'mock' })
agent.send([{ type: 'text', text: 'go' }])
await until(() => gated.started.length === 2)
;(agent as unknown as { currentAbort?: AbortController }).currentAbort?.abort('stop before barrier')
gated.release('1')
gated.release('2')
await waitForIdle(ctx, agent)
expect(exclusive).toEqual([])
expect(events(agent).filter(e => e.type === 'tool/call').map(e => e.data.callId))
.toEqual([CallId('c1'), CallId('c2')])
})
})