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

Merge latest agent execution context base

Browse Source
This commit is contained in:
Yichen Jiang
2026-07-18 21:42:38 +08:00
parent cddb79e844 0e4785cdb7
commit 5190e10d27
105 changed files with 3313 additions and 551 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`)
@@ -59,6 +60,8 @@ Plugin failure ends the current turn, not the loop. The loop creates one private
Cancellation is cooperative: the loop checks for interruption between awaited boundaries but does not abandon an in-process listener, adapter, or tool Promise with `Promise.race`. `whenIdle()` and handle disposal therefore observe real quiescence. See the [explicit turn cancellation RFC](../../../docs/rfc/implemented/architecture/2026-07-16-explicit-turn-cancellation.md).
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:
@@ -85,7 +88,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, normalizeAgentCancelCause } from '@deepseek-ai/dsh-agent'
import type { AgentCancelCause, AgentId, AgentOptions, AgentStatus, InjectOptions, SendOptions } from '@deepseek-ai/dsh-agent'
import type { AgentCancelCause, 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'
@@ -55,15 +55,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 {
@@ -141,19 +142,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
@@ -187,12 +196,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) {
@@ -201,6 +209,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`)
@@ -208,7 +225,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)
@@ -217,7 +234,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)
@@ -233,10 +250,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
@@ -276,6 +298,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(cause?: AgentCancelCause): void {
// Validate before the idle no-op so misuse fails consistently in every state.
const accepted = normalizeAgentCancelCause(cause ?? { kind: 'user' })
@@ -328,6 +378,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) },
installTurnCancellation: () => {
const cancellation = new TurnCancellation()
@@ -342,6 +393,7 @@ export class ReactLoopAgent implements Agent {
isDisposed: () => this._status === 'disposed',
isPreRunCancelled: () => this.preRunCancelled,
clearPreRunCancel: () => { this.preRunCancelled = false },
withToolBatch: run => this.withToolBatch(run),
// Pre-run 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')

62
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'
import type { TurnCancellation } from './cancellation.ts'
@@ -100,6 +101,8 @@ function interruptionTurnEndReason(handle: LoopHandle, signal: AbortSignal): Tur
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
/** Install a fresh active-turn owner before the running notification. */
installTurnCancellation(): TurnCancellation
@@ -114,6 +117,8 @@ export interface LoopHandle {
clearPreRunCancel(): 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>
}
/**
@@ -334,7 +339,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, signal)
ctx, events, agent, handle, turn, step, assembly, fullSystemPrompt, boundaryMessages, transmission, signal)
} catch (error: unknown) {
stepOutcome = { error: toError(error) }
}
@@ -459,6 +464,7 @@ async function runStep(
ctx: Context,
events: AgentEventDispatch,
agent: ReactLoopAgent,
handle: LoopHandle,
turn: number,
step: number,
assembly: PromptAssembly,
@@ -548,53 +554,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) {
interruptionCheckpoint(signal)
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 ?? [])
interruptionCheckpoint(signal)
}
// 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. */

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

@@ -0,0 +1,234 @@
/**
* 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
}
/** Control-flow sentinel; turn classification reads only the explicit signal. */
const TOOL_CALLS_INTERRUPTED = new Error('tool calls interrupted')
/** Stop scheduling at a cooperative boundary without exposing the runtime reason. */
function interruptionCheckpoint(signal: AbortSignal): void {
if (signal.aborted) throw TOOL_CALLS_INTERRUPTED
}
/**
* 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> {
interruptionCheckpoint(signal)
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.
if (signal.aborted) aborted = true
await fillPool()
}
if (aborted) {
// Started calls and accepted context settle before the turn records the abort.
throw TOOL_CALLS_INTERRUPTED
}
/* 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'
@@ -247,6 +247,190 @@ describe('abort during tool execution ends the turn', () => {
expect(adapter.requests).toHaveLength(1) // no follow-up model call
expect(reasons).toEqual([{ kind: 'aborted' }])
})
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.cancel({ kind: 'user' })
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.cancel({ kind: 'user' })
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.cancel({ kind: 'user' })
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', () => {
@@ -631,7 +815,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.cancel({ kind: 'user' })
}
})
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.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(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.cancel({ kind: 'user' })
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.cancel({ kind: 'user' })
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')])
})
})

2
packages/core/agent/README.md
View File

@@ -45,7 +45,7 @@ The handle every plugin programs against:
- `agent.send(content, options?)` — queue a message; starts a turn when idle. Content and resolved source become one detached, deeply frozen lossless-JSON record before `agent/queued` and enqueue; invalid data throws synchronously, and caller or notification-listener in-place mutation cannot change the log or model input (`agent/prompt-submit` still rewrites by returning replacement content).
- `agent.steer(content, options?)` — steer a running turn (inject between steps); uses the same owned acceptance boundary and behaves like `send` when idle
- `agent.inject(content, options?)` — inject in-session context (`context/message` event); the next request sees it. `options.envelope` defaults to the canonical `<context>` framing and may be `'raw'` when the caller owns a complete familiar frame; `options.meta` persists opaque JSON state without rendering it. Does not run the model. While a turn is open it joins that turn; while idle it is wrapped in a one-shot `injection` turn so every event stays turn-enclosed ([the turn-enclosure invariant](../../../docs/rfc/implemented/architecture/2026-06-15-turn-enclosure-invariant.md))
- `agent.inject(content, options?)` — accept detached in-session context without running the model; the next request sees its `context/message`. `options.envelope` defaults to the canonical `<context>` framing and may be `'raw'` when the caller owns a complete familiar frame; `options.meta` persists opaque JSON state without rendering it. While a turn is open it joins that turn, deferring FIFO while the current tool batch executes and draining before turn close if execution is interrupted; while idle it is wrapped in a one-shot `injection` turn and durability checkpoint ([the turn-enclosure invariant](../../../docs/rfc/implemented/architecture/2026-06-15-turn-enclosure-invariant.md)).
- `agent.cancel(cause?)` — cancel ALL pending work: clears the queued + steering FIFOs, aborts the active turn, and drops queued work not yet claimed by the driver. `AgentCancelCause` is the runtime-only `{ kind: 'user' } | { kind: 'parent' }`; omission means `user`, the first cause wins for an active turn, and ACP `session/cancel` maps to `user`. `normalizeAgentCancelCause()` provides the same strict detached-value boundary used by the concrete loop: validation is synchronous even while idle, accepts only an exact plain object, and returns a frozen detached cause. After validation, `agent.cancel()` is a safe no-op when no work exists.
- `agent.whenIdle()` — resolve once the agent reaches quiescence after settling out of `running` (idle → immediately; disposed → awaits the loop exit). A non-owner's quiescence-observation hook: it observes the work settling WITHOUT tearing the agent down. Teardown is separate — a lifecycle owner stops and unregisters via `AgentHandle.dispose()`, which awaits the loop exit directly.
- `agent.session`, `agent.status`, `agent.options`, `agent.id`

9
packages/core/agent/src/types.ts
View File

@@ -183,10 +183,11 @@ export interface Agent {
steer(content: ContentBlock[], options?: SendOptions): void
/**
* Append model-facing context without running the model. Idle injection uses
* a one-shot turn and durability checkpoint, while injection during an open
* turn joins it at the current log position. Disposal awaits idle checkpoints;
* flush failures are reported through `agent/error`, not thrown to the caller.
* Append detached model-facing context without running the model. An open-turn
* injection joins at the current log position unless the current tool batch is
* executing; then it waits FIFO until that batch settles and drains before turn
* close even when interrupted. Idle injection uses a one-shot turn and durability
* checkpoint. Disposal awaits idle checkpoints; flush failures report through `agent/error`.
*/
inject(content: ContentBlock[], options?: InjectOptions): void

80
packages/core/session/tests/gen-persistence-catalog.spec.ts
View File

@@ -9,6 +9,7 @@ import { join } from 'node:path'
import { afterEach, describe, expect, it } from 'vitest'
import {
annotateSurface,
collectEventEnvelopeTypes,
collectLogEvents,
collectSurfaceEventTypes,
render,
@@ -56,6 +57,7 @@ describe('gen-persistence-catalog collectLogEvents', () => {
scope: 'fix',
doc: 'A thing was recorded.',
payload: '{ turn: number }',
declaration: '/** A thing was recorded. */\n\'fix/happened\': { turn: number }',
source: 'packages/core/fix/src/types.ts:3',
})
})
@@ -102,10 +104,13 @@ describe('gen-persistence-catalog collectLogEvents', () => {
it('collapses a newline-separated multi-line payload to a valid one-line fragment', () => {
const events = collectLogEvents(make({
'packages/group/fix/src/types.ts': merge(
' /** Wide payload. */\n \'fix/wide\': {\n alpha: string[]\n range: { start: number; end: number }\n count: number\n }',
' /** Wide payload. */\n \'fix/wide\': {\n /** Alpha values. */\n alpha: string[]\n range: { start: number; end: number }\n count: number\n }',
),
}))
expect(events[0]?.payload).toBe('{ alpha: string[]; range: { start: number; end: number }; count: number }')
expect(events[0]?.declaration).toBe(
'/** Wide payload. */\n\'fix/wide\': {\n /** Alpha values. */\n alpha: string[]\n range: { start: number; end: number }\n count: number\n}',
)
})
it('hard-errors on a member with no description prose', () => {
@@ -158,6 +163,59 @@ describe('gen-persistence-catalog collectLogEvents', () => {
})
})
describe('gen-persistence-catalog collectEventEnvelopeTypes', () => {
const declarations = `/** Event keys. */
export type SessionEventType = keyof SessionEventMap
/** Surface-producing event keys. */
export type SurfaceEventType = 'fix/message'
/** Surface placement. */
export type SurfaceOp = 'append'
/** One persisted event. */
export type SessionEvent<T extends SessionEventType = SessionEventType> = { type: T }
`
it('extracts the envelope declarations with their complete JSDoc in canonical order', () => {
const entries = collectEventEnvelopeTypes(make({
'packages/core/fix/package.json': OWNER_MANIFEST,
'packages/core/fix/src/types.ts': declarations,
}))
expect(entries.map(entry => entry.name)).toEqual([
'SessionEventType',
'SurfaceEventType',
'SurfaceOp',
'SessionEvent',
])
expect(entries[3]).toMatchObject({
declaration: '/** One persisted event. */\nexport type SessionEvent<T extends SessionEventType = SessionEventType> = { type: T }',
source: 'packages/core/fix/src/types.ts:8',
})
})
it('hard-errors when an envelope declaration is missing', () => {
expect(() => collectEventEnvelopeTypes(make({
'packages/core/fix/package.json': OWNER_MANIFEST,
'packages/core/fix/src/types.ts': declarations.replace('/** Surface placement. */\nexport type SurfaceOp = \'append\'\n', ''),
}))).toThrow(/missing event-envelope declaration\(s\): SurfaceOp/)
})
it('hard-errors on duplicate, unexported, undocumented, or mistagged envelope declarations', () => {
const violations = new RegExp([
'4 JSDoc completeness violation\\(s\\)',
'[\\s\\S]*not exported',
'[\\s\\S]*@mode tag',
'[\\s\\S]*SurfaceOp.*no description prose',
'[\\s\\S]*SessionEvent.*already declared',
].join(''))
expect(() => collectEventEnvelopeTypes(make({
'packages/core/fix/package.json': OWNER_MANIFEST,
'packages/core/fix/src/types.ts': declarations
.replace('/** Event keys. */\nexport type SessionEventType', '/** Event keys.\n * @mode emit\n */\ntype SessionEventType')
.replace('/** Surface placement. */\n', '')
+ '/** Duplicate event. */\nexport type SessionEvent = { type: never }\n',
}))).toThrow(violations)
})
})
describe('gen-persistence-catalog collectSurfaceEventTypes', () => {
it('parses the literal union', () => {
const types = collectSurfaceEventTypes(make({
@@ -192,9 +250,21 @@ describe('gen-persistence-catalog annotateSurface + render', () => {
scope: name.split('/')[0] ?? name,
payload: '{ turn: number }',
doc: `Records ${name}.`,
declaration: `/** Records ${name}. */\n'${name}': { turn: number }`,
source: 'packages/core/fix/src/types.ts:3',
})
const envelopeTypes = [
'SessionEventType',
'SurfaceEventType',
'SurfaceOp',
'SessionEvent',
].map(name => ({
name: name as 'SessionEventType' | 'SurfaceEventType' | 'SurfaceOp' | 'SessionEvent',
declaration: `/** ${name}. */\nexport type ${name} = never`,
source: 'packages/core/fix/src/types.ts:1',
}))
it('badges union members surface and everything else log-only', () => {
const annotated = annotateSurface([entry('fix/message'), entry('fix/marker')], ['fix/message'])
expect(annotated.map(e => [e.name, e.surface])).toEqual([['fix/message', true], ['fix/marker', false]])
@@ -205,11 +275,13 @@ describe('gen-persistence-catalog annotateSurface + render', () => {
.toThrow(/'fix\/ghost' name no declared log event/)
})
it('renders badges, payload fences, and the generated-file header', () => {
const out = render(annotateSurface([entry('fix/message'), entry('fix/marker')], ['fix/message']))
it('renders badges, declaration fences, and the generated-file header', () => {
const out = render(annotateSurface([entry('fix/message'), entry('fix/marker')], ['fix/message']), envelopeTypes)
expect(out).toContain('Generated by scripts/gen-persistence-catalog.ts')
expect(out).toContain('# Session Persistence Event Catalog')
expect(out).toContain('```ts persistence-catalog\n/** SessionEventType. */\nexport type SessionEventType = never')
expect(out).toContain('#### `fix/message` — surface')
expect(out).toContain('#### `fix/marker` — log-only')
expect(out).toContain('```ts persistence-catalog\n\'fix/marker\': { turn: number }\n```')
expect(out).toContain('```ts persistence-catalog\n/** Records fix/marker. */\n\'fix/marker\': { turn: number }\n```')
})
})

11
packages/core/tools/README.md
View File

@@ -21,6 +21,7 @@ tools:
- `ctx.tools.schemas(scope?: ScopeKey): ToolSchema[]` Schemas of everything the scope can see (without the `execute` functions). The shipped tools' schemas are catalogued in [docs/tool-catalog.md](../../../docs/tool-catalog.md), generated by booting each tool plugin and harvesting this method (see [the tool-schema-catalog RFC](../../../docs/rfc/implemented/process/2026-07-02-tool-schema-catalog.md)).
- `ctx.tools.guard(guard: ToolGuard): () => void` Register a monotonic synchronous execution guard after `tools/pre-execute`: returning a reason denies the call, while `undefined` leaves it unchanged. A plain-context guard applies globally; an `agent.ctx` guard applies only to that agent. Later waterfall listeners cannot turn a guard denial back into permission. Disposed with the calling fiber.
- `ctx.tools.execute(exec)` losslessly snapshots and freezes arguments, assigns an opaque token, runs the complete policy/dispatch/result pipeline, then independently snapshots the authoritative outcome before final observation. Invalid arguments use the same result path without reaching policy or the body; around wrappers may replace only `signal`.
- `ctx.tools.executionMode(exec)` returns `parallel` only when the visible definition's `isConcurrencySafe(exec.arguments)` classifier returns exactly `true`; unknown, hidden, undeclared, invalid, or throwing classifications are exclusive.
### Injected services
@@ -32,7 +33,7 @@ The live registry pipeline has three transformable waterfalls followed by the ob
### Key types
- `ToolDefinition` — `ToolSchema` + `execute(args, exec)`, optional presentation callbacks, and cooperative `timeoutMs`.
- `ToolDefinition` — `ToolSchema` + `execute(args, exec)`, optional presentation callbacks, cooperative `timeoutMs`, and optional per-call `isConcurrencySafe(args)` classification.
- `ToolExecutionInput` — the caller-supplied call description: `{ callId, name, arguments, agent?, parent?, signal? }`; callers may pass an enclosing execution's opaque token as `parent` but never choose the new execution's own token.
- `ToolExecutionToken` — a fresh branded `Symbol` assigned by the registry. It supports equality correlation only and never crosses a model, log, or worker boundary.
- `ToolExecution` — the pipeline-owned call: immutable `{ token, callId, name, arguments, agent?, parent? }` identity plus optional operational `signal`, which an around wrapper may add, replace, remove, and restore. A nested call's `parent` is a `ToolExecutionToken`, not an execution object.
@@ -87,6 +88,8 @@ See `defineTool`, `validateArgs`, `ToolArgsError`, `SchemaSpec`, `InferArgs`, an
Optional `timeoutMs` must be positive and finite; it is policy metadata, not model-visible schema.
Optional `isConcurrencySafe(args)` receives typed, softly validated arguments. Exact `true` permits concurrent dispatch/body execution; invalid input and all other outcomes remain exclusive. Opted-in bodies do not mutate parent-owned state, and shared-state races must commute or fail closed. The [parallel tool-call RFC](../../../docs/rfc/implemented/feature/2026-07-10-parallel-tool-call-execution.md) owns the full safety contract.
### Structured-output schema subset
`StructuredOutputSchema` is the object-rooted raw JSON Schema subset used by subagents and workflows for machine-readable results. It accepts one scalar `type`, object `properties`/`required`/boolean `additionalProperties`, array `items`, and scalar `enum`/`const`. The annotations `description`, `title`, `default`, and `examples` are ignored but must remain JSON data. Type arrays, undeclared required keys, and unsupported keywords fail through `OutputSchemaError` rather than being ignored; `validateStructuredValue()` returns path-qualified violations without throwing.
@@ -108,6 +111,10 @@ Under `code` or `both`, the registry exposes the reserved `run_code` transport a
- **The dispatch bridge** (`run_code`'s execute): every binding call is JSON-normalized before dispatch (a value that does not survive — `BigInt`, circulars — rejects that one call, so the dispatched form and logged form are the same JSON value by construction), serialized through a per-run queue (even `Promise.all` executes underlying calls one at a time in submission order), given the outer execution's opaque token as `parent`, and run through the complete pre-execute → guards → execute → post-execute → result pipeline. A denial reaches the program as a binding rejection, and each sub-call is logged as a `tool/code-dispatch` session event with deterministic id `<parent>:code:<n>`; `deriveMessages()` does not surface that event. Token correlation lets commit-style observers defer an inner success until the final `run_code` result without exposing the live outer execution; ordinary tool side effects are not rolled back. Every sub-call `additionalContexts` entry is deferred through the outer `ToolRunContext` in dispatch order; the loop appends those contexts only after the parent `run_code` result, preserving adjacency and retaining each source/envelope/meta even when the program later fails.
- **Settlement discipline**: the bridge owns a run-scoped abort that follows the outer signal in and fires when the run settles for any reason, so a budget expiry aborts an in-flight sub-tool instead of orphaning it; the bridge then drains its queue BEFORE returning, so every `tool/code-dispatch` lands inside the open turn. A failed run throws `CodeRunFailedError` (`code: 'CODE_RUN_FAILED'`, message = the failure kind + captured logs), which the pipeline converts to a structured `isError` the model self-corrects from.
### Parallel execution
The agent loop groups consecutive `parallel` calls into a bounded rolling pool and treats each `exclusive` call as an ordering barrier. Only dispatch/body overlaps; policy, durable results, and context retain model order. Code Mode bindings remain serial. The [parallel tool-call RFC](../../../docs/rfc/implemented/feature/2026-07-10-parallel-tool-call-execution.md) owns the shipped declarations and rationale.
## Model Experience
### Normal tool schemas
@@ -145,7 +152,7 @@ The available tools:
## Known Limitations and Deferred Work
- **Native tool calls execute sequentially** — `ToolDefinition` carries no concurrency-safety metadata; adding it (and parallel execution in the loop) waits on the deferred tool-shapes review (`TODO(review)`).
- **Concurrency policy is not an event seam** — `executionMode()` reads the resolved tool definition directly; plugins can only declare a classifier on definitions they own.
- **`tools/pre-execute` deliberately cannot rewrite `exec.arguments`** — logged and rendered args would desync from what ran; the rewrite design is [a proposed RFC](../../../docs/rfc/proposed/feature/2026-06-30-pre-tool-input-rewrite.md).
- **`defineTool`'s schema DSL is a deliberate subset** — string/number/boolean/object/array with string-only `enum`; `validateArgs` tolerates extra keys and never applies `default` (`XXX(unused-default)` flags removing that field); raw-registered JSON-Schema tools validate their own input.
- **`timeoutMs` on a definition is declarative only** — the registry never enforces deadlines; enforcement requires the `@deepseek-ai/dsh-timeout-policy` wrapper.

326
packages/core/tools/src/index.ts
View File

@@ -117,9 +117,6 @@ declare module 'cordis' {
}
}
// TODO(concurrency): revisit these shapes when concurrency metadata becomes useful
// (for example, a read-only hint that would permit safe parallel execution).
/** Tool output, optionally with lossless-JSON presentation metadata persisted for replay. */
export type ToolExecuteReturn = ContentBlock[] | { content: ContentBlock[]; meta?: unknown }
@@ -134,6 +131,20 @@ export interface ToolDefinition extends ToolSchema {
* cooperative implementation that can reach quiescence when the signal aborts.
*/
timeoutMs?: number
/**
* Pure synchronous classifier for overlap with sibling tool calls. Only
* `true` opts in; omission, exceptions, non-`true` returns, and invalid
* `defineTool` arguments are exclusive. This metadata is never model-visible.
*
* Opted-in executions must not mutate parent-owned state. Shared state must
* tolerate concurrent dispatch; recorder races are permitted only when they
* commute or fail closed. See the
* [parallel-tool-call RFC](../../../../docs/rfc/implemented/feature/2026-07-10-parallel-tool-call-execution.md)
* for the full contract.
* @param args - parsed arguments; `defineTool` validates before calling.
* @returns Whether this call may join a parallel group.
*/
isConcurrencySafe?(args: unknown): boolean
/**
* Optional: how to present the PENDING state of one call in a UI, derived from
* the call's `args` (parsed arguments, `unknown` — the tool validates/narrows
@@ -195,6 +206,14 @@ export interface ToolExecutionInput {
signal?: AbortSignal
}
/**
* Scheduling mode for one pending call. `parallel` may overlap with siblings;
* `exclusive` runs alone and forms an ordering barrier.
*/
export type ToolExecutionMode =
| { kind: 'parallel' }
| { kind: 'exclusive' }
/**
* One pending tool call inside the registry pipeline. Parsed arguments cross
* one lossless-JSON materialization boundary before policy and are deep-frozen;
@@ -222,6 +241,47 @@ export interface ToolRunContext extends ToolExecution {
deferContext(context: HookContext): void
}
/**
* Scheduler-only result after ordered pre-execute and guards. A `post-result`
* still receives post-execute; a `final-result` bypasses it.
* @internal
*/
export type ScheduledToolPreparation =
| { kind: 'dispatch'; exec: ToolRunContext }
| { kind: 'post-result'; exec: ToolRunContext; result: ToolExecutionResult }
| { kind: 'final-result'; exec: ToolRunContext; result: ToolExecutionResult }
/**
* Scheduler-only dispatch result. A `post-result` still receives post-execute;
* a `final-result` already matches {@link ToolRegistry.execute} failure semantics.
* @internal
*/
export type ScheduledToolDispatch =
| { kind: 'post-result'; result: ToolExecutionResult }
| { kind: 'final-result'; result: ToolExecutionResult }
/**
* Symbol-keyed scheduler view that keeps pre/post policy ordered while
* overlapping dispatch. Ordinary callers use {@link ToolRegistry.execute};
* this is not a plugin seam.
* @internal
*/
export interface ToolRegistryScheduler {
/** Materialize input, run the ordered pre-execute/guard gate, and decide what stage follows. */
prepare(exec: ToolExecutionInput): Promise<ScheduledToolPreparation>
/** Run only the around-dispatch/body stage. */
dispatch(exec: ToolRunContext): Promise<ScheduledToolDispatch>
/** Run ordered post-execute finalization, then materialize and notify the final outcome. */
finalize(exec: ToolRunContext, result: ToolExecutionResult): Promise<ToolExecutionResult>
/** Materialize and notify a final outcome that must bypass post-execute. */
finish(exec: ToolRunContext, result: ToolExecutionResult): ToolExecutionResult
}
/**
* Scheduler entry point omitted from the generated named service API.
* @internal
*/
export const TOOL_REGISTRY_SCHEDULER: unique symbol = Symbol('@deepseek-ai/dsh-tools.scheduler')
/** Structured error metadata for a failed tool call (alongside the model-facing text). */
export interface ToolErrorInfo {
name: string
@@ -252,8 +312,8 @@ export interface ToolExecutionResult {
*/
error?: ToolErrorInfo
/**
* Model-facing context for the next request, separate from this tool result.
* The loop buffers it until all step results are logged, preserving pairing.
* Model-facing context for the next request, separate from this tool result. The loop
* accepts it into the active-batch FIFO, then appends after recorded results even if interrupted.
*/
additionalContexts?: HookContext[]
/**
@@ -382,6 +442,16 @@ export class ToolRegistry extends Service {
mode: z.union(['native', 'code', 'both'] as const).default('native'),
})
/** Internal staged view consumed by `dsh-agent-loop`'s parallel scheduler. */
readonly [TOOL_REGISTRY_SCHEDULER]: ToolRegistryScheduler = {
prepare: exec => this.prepareScheduledExecution(exec),
dispatch: exec => this.dispatchScheduledExecution(exec),
finalize: (exec, result) => this.finalizeScheduledExecution(exec, result),
finish: (exec, result) => this.finishScheduledExecution(exec, result),
}
/** Context deferred by a running tool body, keyed by its scheduler-owned execution. */
private deferredContexts = new WeakMap<ToolRunContext, HookContext[]>()
private global = new Map<string, ToolDefinition>()
private scoped = new Map<ScopeKey, Map<string, ToolDefinition>>()
/** Compiled restriction filters, per scope (see {@link restrict}). */
@@ -682,6 +752,24 @@ export class ToolRegistry extends Service {
}
}
/**
* Classify a pending call through the caller's visible tool definition. Only
* an exact `true` is parallel; unknown, hidden, undeclared, invalid, or
* throwing classifiers are exclusive.
* @param exec - call name, parsed arguments, and optional agent scope.
* @returns the fail-closed scheduling mode.
*/
executionMode(exec: ToolExecutionInput): ToolExecutionMode {
const tool = this.get(exec.name, exec.agent)
if (!tool?.isConcurrencySafe) return { kind: 'exclusive' }
try {
const concurrencySafe: unknown = tool.isConcurrencySafe(exec.arguments)
return concurrencySafe === true ? { kind: 'parallel' } : { kind: 'exclusive' }
} catch {
return { kind: 'exclusive' }
}
}
/**
* Execute through pre-policy, guards, around-dispatch, post-policy, and final
* notification. Tool and listener failures resolve as materialized error
@@ -692,6 +780,28 @@ export class ToolRegistry extends Service {
* @returns the materialized final result.
*/
async execute(exec: ToolExecutionInput): Promise<ToolExecutionResult> {
return this.prepareExecution(exec, prepared => this.completeScheduledExecution(prepared))
}
private async completeScheduledExecution(prepared: ScheduledToolPreparation): Promise<ToolExecutionResult> {
switch (prepared.kind) {
case 'dispatch': {
const dispatched = await this.dispatchScheduledExecution(prepared.exec)
return dispatched.kind === 'post-result'
? await this.finalizeScheduledExecution(prepared.exec, dispatched.result)
: this.finishScheduledExecution(prepared.exec, dispatched.result)
}
case 'post-result':
return await this.finalizeScheduledExecution(prepared.exec, prepared.result)
case 'final-result':
return this.finishScheduledExecution(prepared.exec, prepared.result)
/* v8 ignore next -- closed-union exhaustiveness guard */
default:
return assertNever(prepared, 'scheduled tool preparation')
}
}
private createExecution(exec: ToolExecutionInput): ScheduledToolPreparation | { kind: 'ready'; exec: ToolRunContext } {
const deferredContexts: HookContext[] = []
const token = createExecutionToken()
const callId = exec.callId
@@ -710,105 +820,143 @@ export class ToolRegistry extends Service {
deferredContexts.push(context)
},
}
let execution: ToolRunContext
try {
const detached = snapshotJsonValue(exec.arguments)
if (detached === undefined) {
throw new TypeError('tool execution arguments must be losslessly JSON-serializable')
}
execution = {
...base,
arguments: deepFreeze(detached),
}
const execution: ToolRunContext = { ...base, arguments: deepFreeze(detached) }
this.deferredContexts.set(execution, deferredContexts)
return { kind: 'ready', exec: execution }
} catch (error: unknown) {
execution = { ...base, arguments: undefined }
const result = this.materializeFinalResult(toolErrorResult(error))
this.notifyResult(execution, result)
return result
const execution: ToolRunContext = { ...base, arguments: undefined }
return { kind: 'final-result', exec: execution, result: toolErrorResult(error) }
}
let result: ToolExecutionResult
}
/**
* Run the ordered pre-execute and monotonic guard stages for the scheduler.
* @param input - the caller-supplied execution input.
* @returns the prepared execution plus the next scheduler stage.
* @internal
*/
private async prepareScheduledExecution(input: ToolExecutionInput): Promise<ScheduledToolPreparation> {
return this.prepareExecution(input, prepared => prepared)
}
private async prepareExecution<T>(
input: ToolExecutionInput,
next: (prepared: ScheduledToolPreparation) => T | PromiseLike<T>,
): Promise<T> {
const created = this.createExecution(input)
if (created.kind !== 'ready') return next(created)
const exec = created.exec
try {
result = this.materializeFinalResult(await this.executePipeline(execution, deferredContexts))
const carrier = scopeTarget(this, exec.agent)
const gate = await this.ctx.waterfall(
carrier, 'tools/pre-execute', exec,
() => Promise.resolve<PreToolDecision>({ kind: 'allow' }),
)
const decision = gate.kind === 'ask' ? await this.serviceAsk(exec, gate) : gate
const denialReason = decision.kind === 'allow'
? this.guardReason(exec)
: decision.reason
if (denialReason !== undefined) {
return await next({
kind: 'post-result',
exec,
result: {
content: [{ type: 'text', text: `Error: ${denialReason}` }],
isError: true,
},
})
}
return await next({ kind: 'dispatch', exec })
} catch (error: unknown) {
// Outer backstop: a throwing pre/post-execute listener, guard, or the
// waterfall machinery becomes an isError result, never a turn failure.
result = this.materializeFinalResult(toolErrorResult(error))
return next({ kind: 'final-result', exec, result: toolErrorResult(error) })
}
this.notifyResult(execution, result)
return result
}
/** Run the transformable pipeline; {@link execute} owns final normalization and notification. */
private async executePipeline(exec: ToolRunContext, deferredContexts: HookContext[]): Promise<ToolExecutionResult> {
// --- Gate: tools/pre-execute. An `ask` resolves through the optional
// approval seam (or degrades to deny) before the monotonic guards run. The
// carrier keys dispatch by exec.agent, so an `agent.ctx` listener gates only
// its own agent's calls (agent-less calls are subject-less).
const carrier = scopeTarget(this, exec.agent)
const gate = await this.ctx.waterfall(
carrier, 'tools/pre-execute', exec,
() => Promise.resolve<PreToolDecision>({ kind: 'allow' }),
)
const decision = gate.kind === 'ask' ? await this.serviceAsk(exec, gate) : gate
const denialReason = decision.kind === 'allow'
? this.guardReason(exec)
: decision.reason
if (denialReason !== undefined) {
// Every non-grant, including a failed/unavailable approval request, takes
// the same deny path and still reaches post-policy plus result observers.
const denied: ToolExecutionResult = {
content: [{ type: 'text', text: `Error: ${denialReason}` }],
isError: true,
}
return await this.postExecute(exec, denied)
}
// --- Around-dispatch: tools/execute. The base `next` is the dispatch-
// with-normalization thunk — the tool body's own try/catch turns a throw
// into an isError result so a wrapper (and post-execute) can inspect it;
// an unknown tool routes through the same catch. A `tools/execute` listener
// (e.g. a timeout plugin) wraps this thunk: it may replace `exec.signal`
// before delegating and inspect the normalized result after. Dispatched with the
// same carrier as the gate, so an `agent.ctx` wrapper wraps only its own
// agent's calls. ---
const result = await this.ctx.waterfall(
carrier, 'tools/execute', exec,
async (): Promise<ToolExecutionResult> => {
try {
// Resolve through the CALLER's visible view ({@link get}): a scoped
// tool shadows its global name-twin for that agent, and a
// restricted-away global tool is exactly as absent as a nonexistent
// one — same UNKNOWN_TOOL result, no capability leak in the error.
const tool = this.get(exec.name, exec.agent)
if (!tool) throw new ToolNotFoundError(exec.name)
// Normalize the two `execute` return shapes: a bare ContentBlock[] (no
// meta) or a { content, meta } object (a tool attaching a private
// presentation payload). An array IS the content; the object carries it.
const returned = await tool.execute(exec.arguments, exec)
const content = Array.isArray(returned) ? returned : returned.content
const meta = Array.isArray(returned) ? undefined : returned.meta
return { content, isError: false, ...meta !== undefined ? { meta } : {} }
} catch (error: unknown) {
return toolErrorResult(error)
/**
* Run around-dispatch and the tool body. Tool and unknown-tool failures still
* receive post-execute; pipeline failures are already final.
* @param exec - the prepared execution.
* @returns whether the result still needs post-execute.
* @internal
*/
private async dispatchScheduledExecution(exec: ToolRunContext): Promise<ScheduledToolDispatch> {
try {
const carrier = scopeTarget(this, exec.agent)
const result = await this.ctx.waterfall(
carrier, 'tools/execute', exec,
async (): Promise<ToolExecutionResult> => {
try {
const tool = this.get(exec.name, exec.agent)
if (!tool) throw new ToolNotFoundError(exec.name)
const returned = await tool.execute(exec.arguments, exec)
const content = Array.isArray(returned) ? returned : returned.content
const meta = Array.isArray(returned) ? undefined : returned.meta
return { content, isError: false, ...meta !== undefined ? { meta } : {} }
} catch (error: unknown) {
return toolErrorResult(error)
}
},
)
const deferredContexts = this.deferredContexts.get(exec)
/* v8 ignore next -- dispatch only receives executions minted by this registry's prepare stage */
if (deferredContexts === undefined) throw new Error('tool registry scheduler invariant violated: unprepared execution')
const resultWithDeferredContexts: ToolExecutionResult = deferredContexts.length === 0
? result
: {
...result,
additionalContexts: [
...deferredContexts,
...result.additionalContexts ?? [],
],
}
},
)
const resultWithDeferredContexts: ToolExecutionResult = deferredContexts.length === 0
? result
: {
...result,
additionalContexts: [
...deferredContexts,
...result.additionalContexts ?? [],
],
}
return await this.postExecute(exec, resultWithDeferredContexts)
return { kind: 'post-result', result: resultWithDeferredContexts }
} catch (error: unknown) {
return { kind: 'final-result', result: toolErrorResult(error) }
}
}
/** Notify final-result observers without giving them a mutation/error channel into the outcome. */
/**
* Run ordered post-execute, then materialize and notify the final outcome.
* @param exec - the prepared execution.
* @param result - dispatch/pre result that still needs post-execute.
* @returns the materialized final result.
* @internal
*/
private async finalizeScheduledExecution(exec: ToolRunContext, result: ToolExecutionResult): Promise<ToolExecutionResult> {
try {
return this.finishScheduledExecution(exec, await this.postExecute(exec, result))
} catch (error: unknown) {
return this.finishScheduledExecution(exec, toolErrorResult(error))
}
}
/**
* Materialize and notify a final result that must bypass post-execute.
* @param exec - the prepared execution.
* @param result - final result.
* @returns the materialized final result.
* @internal
*/
private finishScheduledExecution(exec: ToolRunContext, result: ToolExecutionResult): ToolExecutionResult {
let finalResult: ToolExecutionResult
try {
finalResult = this.materializeFinalResult(result)
} catch (error: unknown) {
finalResult = this.materializeFinalResult(toolErrorResult(error))
}
this.notifyResult(exec, finalResult)
return finalResult
}
/** Notify observers without exposing a mutation or error channel into the outcome. */
private notifyResult(exec: ToolExecution, result: ToolExecutionResult): void {
// The pipeline is over: freeze the remaining mutable signal slot so every
// observer sees the SAME WeakMap-keyable execution without a mutation race.
// Freeze the remaining mutable signal slot before observers receive the
// shared WeakMap-keyable execution object.
Object.freeze(exec)
const callbacks = this.ctx.events.dispatch('emit', [
scopeTarget(this, exec.agent), 'tools/result', exec, result,
@@ -865,7 +1013,7 @@ export class ToolRegistry extends Service {
* its {@link PostToolDecision}: `accept` keeps the call successful (replacing
* `content` when given), `block` turns it into an `isError` whose content is
* the corrective `feedback`. Either decision may attach `additionalContexts`,
* which are ferried on the returned result for the loop's per-step buffer.
* which are ferried on the returned result for the loop's active-batch FIFO.
* Context deferred by the tool body survives an accepted result but is
* discarded when the outer call is blocked; a block exposes only context the
* blocking decision explicitly supplied.

19
packages/core/tools/src/schema.ts
View File

@@ -283,6 +283,14 @@ export interface DefineToolOptions<S extends SchemaSpec> {
* is never sent to the model.
*/
readonly timeoutMs?: number
/**
* Optional pure synchronous classifier for sibling overlap. It receives typed
* arguments after soft validation; invalid input returns `false` without
* invoking it. See {@link ToolDefinition.isConcurrencySafe}.
* @param args - typed validated arguments.
* @returns whether this call may join a parallel group.
*/
isConcurrencySafe?(args: InferArgs<S>): boolean
/**
* Tool execution function. `args` is typed as {@link InferArgs<S>} — zero
* casts needed. Returns either a bare {@link ContentBlock}`[]` (model-facing
@@ -315,7 +323,7 @@ export interface DefineToolOptions<S extends SchemaSpec> {
* @param options - the tool's name, description, typed parameter schema,
* execute body, and optional presenters.
* @returns a registry-ready definition with strict execution validation and
* soft presenter validation for replay compatibility.
* soft presenter and classifier validation for replay compatibility.
*/
export function defineTool<S extends SchemaSpec>(options: DefineToolOptions<S>): ToolDefinition {
// Object-literal execute methods don't use `this`; the reference is safe.
@@ -325,6 +333,8 @@ export function defineTool<S extends SchemaSpec>(options: DefineToolOptions<S>):
const userPresentCall = options.presentCall
// eslint-disable-next-line @typescript-eslint/unbound-method
const userPresentResult = options.presentResult
// eslint-disable-next-line @typescript-eslint/unbound-method
const userIsConcurrencySafe = options.isConcurrencySafe
if (options.timeoutMs !== undefined && (!Number.isFinite(options.timeoutMs) || options.timeoutMs <= 0)) {
throw new Error(`defineTool(${options.name}): timeoutMs must be a positive finite number`)
}
@@ -359,5 +369,12 @@ export function defineTool<S extends SchemaSpec>(options: DefineToolOptions<S>):
return userPresentResult(args as InferArgs<S>, result)
}
}
// Invalid arguments fail closed without invoking the typed classifier.
if (userIsConcurrencySafe) {
tool.isConcurrencySafe = (args: unknown): boolean => {
if (validateArgs(options.parameters, args).length > 0) return false
return userIsConcurrencySafe(args as InferArgs<S>)
}
}
return tool
}

136
packages/core/tools/tests/execution-mode.spec.ts Normal file
View File

@@ -0,0 +1,136 @@
/** Covers fail-closed per-call classification and model-schema isolation. */
import { describe, expect, expectTypeOf, it } from 'vitest'
import { Context } from 'cordis'
import { CallId } from '@deepseek-ai/dsh-llm'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry, {
defineTool,
type ToolDefinition,
type ToolExecutionInput,
type ToolExecutionMode,
} from '@deepseek-ai/dsh-tools'
async function setup() {
const ctx = new Context()
await ctx.plugin(SystemPrompt)
await ctx.plugin(ToolRegistry)
return ctx
}
function exec(name: string, args: unknown): ToolExecutionInput {
return { callId: CallId('c1'), name, arguments: args }
}
describe('ToolRegistry.executionMode', () => {
it('returns parallel only for an explicit true classifier', async () => {
const ctx = await setup()
ctx.tools.register(defineTool({
name: 'safe',
description: 'parallel-safe',
parameters: {},
isConcurrencySafe: () => true,
async execute() { return [] },
}))
expect(ctx.tools.executionMode(exec('safe', {}))).toEqual({ kind: 'parallel' })
})
it('defaults to exclusive for a tool with no isConcurrencySafe declaration', async () => {
const ctx = await setup()
ctx.tools.register(defineTool({
name: 'plain',
description: 'no declaration',
parameters: {},
async execute() { return [] },
}))
expect(ctx.tools.executionMode(exec('plain', {}))).toEqual({ kind: 'exclusive' })
})
it('returns exclusive for an unknown tool', async () => {
const ctx = await setup()
expect(ctx.tools.executionMode(exec('nonexistent', {}))).toEqual({ kind: 'exclusive' })
})
it('returns exclusive when the classifier returns false for these args', async () => {
const ctx = await setup()
ctx.tools.register(defineTool({
name: 'rw',
description: 'read or write',
parameters: { mode: { type: 'string', required: true } },
isConcurrencySafe: args => args.mode === 'read',
async execute() { return [] },
}))
expect(ctx.tools.executionMode(exec('rw', { mode: 'read' }))).toEqual({ kind: 'parallel' })
expect(ctx.tools.executionMode(exec('rw', { mode: 'write' }))).toEqual({ kind: 'exclusive' })
})
it('classifies invalid defineTool arguments as exclusive without throwing', async () => {
const ctx = await setup()
ctx.tools.register(defineTool({
name: 'needs-mode',
description: 'requires mode',
parameters: { mode: { type: 'string', required: true } },
isConcurrencySafe: () => true,
async execute() { return [] },
}))
expect(ctx.tools.executionMode(exec('needs-mode', {}))).toEqual({ kind: 'exclusive' })
})
it('treats a throwing raw classifier as exclusive', async () => {
const ctx = await setup()
const raw: ToolDefinition = {
name: 'thrower',
description: 'classifier throws',
parameters: { type: 'object', properties: {} },
isConcurrencySafe() { throw new Error('boom') },
async execute() { return [] },
}
ctx.tools.register(raw)
expect(ctx.tools.executionMode(exec('thrower', {}))).toEqual({ kind: 'exclusive' })
})
it('treats a truthy non-boolean raw result as exclusive', async () => {
const ctx = await setup()
const raw = {
name: 'truthy',
description: 'classifier returns a truthy string',
parameters: { type: 'object', properties: {} },
isConcurrencySafe() { return 'yes' },
async execute() { return [] },
} as unknown as ToolDefinition
ctx.tools.register(raw)
expect(ctx.tools.executionMode(exec('truthy', {}))).toEqual({ kind: 'exclusive' })
})
it('passes parsed arguments directly to a raw definition', async () => {
const ctx = await setup()
let seen: unknown
ctx.tools.register({
name: 'raw-safe',
description: 'raw',
parameters: { type: 'object', properties: {} },
isConcurrencySafe(args) { seen = args; return true },
async execute() { return [] },
})
expect(ctx.tools.executionMode(exec('raw-safe', { anything: 1 }))).toEqual({ kind: 'parallel' })
expect(seen).toEqual({ anything: 1 })
})
it('isConcurrencySafe never reaches the model-facing schemas() projection', async () => {
const ctx = await setup()
ctx.tools.register(defineTool({
name: 'safe',
description: 'parallel-safe',
parameters: { x: { type: 'string', required: true } },
isConcurrencySafe: () => true,
async execute() { return [] },
}))
const schema = ctx.tools.schemas()[0] as unknown as Record<string, unknown>
expect(Object.keys(schema).sort()).toEqual(['description', 'name', 'parameters'])
expect(schema.isConcurrencySafe).toBeUndefined()
})
it('ToolExecutionMode is the object-tagged union', () => {
expectTypeOf<ToolExecutionMode>().toEqualTypeOf<{ kind: 'parallel' } | { kind: 'exclusive' }>()
})
})