feat(agent-loop): run safe tool calls in parallel
This commit is contained in:
Dudu-0223
@@ -26,14 +26,15 @@ The config-driven `ctx.agentLoop.create()` path keeps its agent owned by the loo
|
||||
```ts
|
||||
interface Config {
|
||||
agents: Array<{
|
||||
id: string // required
|
||||
id: string // required
|
||||
model?: string
|
||||
cwd?: string // optional workspace cwd for the fresh session
|
||||
cwd?: string // optional workspace cwd for the fresh session
|
||||
maxParallelToolCalls?: number // positive integer; per-agent parallel tool-call cap (default 10)
|
||||
}>
|
||||
}
|
||||
```
|
||||
|
||||
Agents listed in config are auto-created at startup. `cwd` applies only to fresh config-created sessions; `resumeSessionId` keeps the persisted session header. There is no per-agent persona: the deployment persona is `dsh-system-prompt`'s own `persona` config, shared by every agent in the context. The plugin registers the built-in `model`/`cwd` prompt variables on `ctx.systemPrompt`, resolved per step from the `assemble({ agent })` context — runtime facts of the agents THIS loop drives, unlike the `harness:identity`/`deployment:persona` sections, which live on `dsh-system-prompt` so they survive a swapped loop plugin.
|
||||
Agents listed in config are auto-created at startup. `cwd` applies only to fresh config-created sessions; `resumeSessionId` keeps the persisted session header. `maxParallelToolCalls` (a positive integer, default `DEFAULT_MAX_PARALLEL_TOOL_CALLS` = `10`) bounds how many parallel-safe calls one assistant step runs at once; `1` restores fully serial execution. It is validated in the schema (`z.number().step(1).min(1)`), so a bad value fails config load rather than being silently dropped. There is no per-agent persona: the deployment persona is `dsh-system-prompt`'s own `persona` config, shared by every agent in the context. The plugin registers the built-in `model`/`cwd` prompt variables on `ctx.systemPrompt`, resolved per step from the `assemble({ agent })` context — runtime facts of the agents THIS loop drives, unlike the `harness:identity`/`deployment:persona` sections, which live on `dsh-system-prompt` so they survive a swapped loop plugin.
|
||||
|
||||
### Classes
|
||||
|
||||
@@ -67,10 +68,12 @@ forever:
|
||||
stream llm.stream(freeze({header..., messages: prefix+boundary})) → session('assistant/chunk')
|
||||
message = waterfall agent/step-result
|
||||
session('assistant/message')
|
||||
each tool-call: session('tool/call')
|
||||
→ tools.execute() [waterfall tools/pre-execute → dispatch → tools/post-execute]
|
||||
→ session('tool/result')
|
||||
append buffered post-execute additionalContext as session('context/message')(s)
|
||||
schedule tool-calls: group by tools.executionMode (exclusive call = barrier;
|
||||
run of parallel-safe calls = one rolling-pool group, ≤ maxParallelToolCalls in flight)
|
||||
each STARTED call: session('tool/call') ⟵ model-order per started call; log positions
|
||||
→ ordered tools/pre-execute → pooled dispatch/body → ordered tools/post-execute may interleave with sibling results as the pool replenishes
|
||||
commit cursor appends session('tool/result') in MODEL order (slot-buffered)
|
||||
append buffered post-execute additionalContext (model call order) as session('context/message')(s)
|
||||
drain steering → session('steering/message')
|
||||
cont = waterfall agent/turn-continuation → ContinuationDecision
|
||||
({action:'continue', reason?} records reason as next-step steering)
|
||||
@@ -83,6 +86,8 @@ forever:
|
||||
|
||||
Error containment: a throwing plugin ends the **turn**, never the loop. Dispose mid-turn emits `agent/status('disposed')` and ends with reason `disposed`. A step that hits the model's output-token ceiling makes the turn end `max-tokens` (the rule: any `max-tokens` step in the turn surfaces as `max-tokens`; `disposed`/`aborted`/`error` still take precedence) — distinct from a clean `completed` stop.
|
||||
|
||||
Tool scheduling: within one assistant step the loop partitions tool calls into ordered groups via `ctx.tools.executionMode` — an exclusive call is its own group (an ordering barrier), a run of consecutive parallel-safe calls is one group. A parallel group runs in a rolling pool: up to `maxParallelToolCalls` calls start in model order, and each settle starts the next until the group drains. Only dispatch/body overlaps — `tools/pre-execute`/`tools/post-execute` run in model call order, each STARTED call appends its own `tool/call` (whose log position may interleave with sibling `tool/result`s), and a model-order commit cursor appends `tool/result` from slot-buffered settlements so derived history stays model-ordered (pairing by the assistant message + `callId`). `additionalContext` from the group is injected in model call order after every result. Abort stops replenishment, drains only already-started calls to results, drops buffered context, and re-raises so `runTurn` owns the end reason; a group not yet started appends no `tool/call`. `maxParallelToolCalls: 1` is byte-for-byte the old serial path.
|
||||
|
||||
Cancellation: `agent.cancel()` is the single public stop primitive — it clears the queued + steering FIFOs, aborts the in-flight step, and drives a turn-scoped marker the driver checks at every point a turn could start or continue (right after the idle wait, after the `running` flip, before each step, and at the continuation gate) so a turn about to start is dropped. A cancelled turn ends `aborted`; a queued-but-not-started prompt never runs and cannot be batched into the cancelled turn. The marker is reset once per loop iteration, so a cancel governs exactly one turn and never leaks onto a later prompt. (The loop still aborts its own per-step `AbortController` directly on disposal and from `cancel()`; that controller is loop-internal, not a public verb.)
|
||||
|
||||
### What is NOT here
|
||||
|
||||
15
packages/core/agent-loop/src/constants.ts
Normal file
15
packages/core/agent-loop/src/constants.ts
Normal file
@@ -0,0 +1,15 @@
|
||||
/**
|
||||
* Loop-level tunable defaults shared between the plugin entry (`index.ts`) and
|
||||
* the tool-call scheduler (`tool-calls.ts`). Kept in a leaf module so importing
|
||||
* a default never pulls in the service class or the scheduler.
|
||||
*
|
||||
* @module dsh-agent-loop/constants
|
||||
*/
|
||||
|
||||
/**
|
||||
* Default cap on simultaneously in-flight tool calls within one assistant step,
|
||||
* when {@link AgentOptions.maxParallelToolCalls} is unset. Matches the
|
||||
* rolling-pool size Claude Code uses; a group larger than the cap is not
|
||||
* truncated — the cap limits concurrency, not the group.
|
||||
*/
|
||||
export const DEFAULT_MAX_PARALLEL_TOOL_CALLS = 10
|
||||
@@ -29,6 +29,22 @@ declare module 'cordis' {
|
||||
}
|
||||
}
|
||||
|
||||
declare module '@deepseek-ai/dsh-agent' {
|
||||
interface AgentOptions {
|
||||
/**
|
||||
* Maximum tool calls this agent runs concurrently within one assistant step
|
||||
* (a positive integer; defaults to {@link DEFAULT_MAX_PARALLEL_TOOL_CALLS}).
|
||||
* The loop's rolling pool starts up to this many parallel-safe calls at once
|
||||
* and replenishes as each settles; `1` preserves the fully serial path.
|
||||
* A merge-extensible field — the loop owns it (it neither the agent nor the
|
||||
* subagent seam sets it), read in `runStep` when scheduling a parallel group.
|
||||
*/
|
||||
maxParallelToolCalls?: number
|
||||
}
|
||||
}
|
||||
|
||||
export { DEFAULT_MAX_PARALLEL_TOOL_CALLS } from './constants.ts'
|
||||
|
||||
/**
|
||||
* Plugin config: the agents to create — or resume, via `resumeSessionId` —
|
||||
* declaratively at startup, so a cordis.yml deployment needs no code.
|
||||
@@ -40,6 +56,11 @@ export interface Config {
|
||||
id: AgentId
|
||||
/** Optional workspace cwd for the config-created fresh session. */
|
||||
cwd?: string
|
||||
/**
|
||||
* Maximum parallel-safe tool calls to run concurrently within one assistant
|
||||
* step. Must be a positive integer; `1` preserves serial execution.
|
||||
*/
|
||||
maxParallelToolCalls?: number
|
||||
/**
|
||||
* If set, the config agent RESUMES this persisted session id instead of
|
||||
* starting a fresh `${id}-session-<uuid>`. Sourced from an env var in
|
||||
@@ -81,6 +102,9 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
model: z.string(),
|
||||
cwd: z.string(),
|
||||
resumeSessionId: z.string(),
|
||||
// A positive integer; a bad value (0, negative, fractional) fails config
|
||||
// validation here rather than being silently dropped from cordis.yml.
|
||||
maxParallelToolCalls: z.number().step(1).min(1),
|
||||
})).default([]),
|
||||
}) as unknown as z<Config>
|
||||
|
||||
@@ -144,6 +168,7 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
* @returns the running agent, owned by the calling fiber (no handle).
|
||||
*/
|
||||
create(id: AgentId, options: AgentOptions = {}, meta: Pick<SessionHeader, 'cwd'> = {}): ReactLoopAgent {
|
||||
this.validateAgentOptions(options)
|
||||
this.assertAgentIdFree(id)
|
||||
// Config/programmatic path: prepare the session and let start() fold its
|
||||
// lifecycle into the agent's composite effect (so a fiber unload tears the
|
||||
@@ -168,6 +193,7 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
* @returns the handle whose dispose tears down exactly this agent.
|
||||
*/
|
||||
createAgent(options: CreateAgentOptions): AgentHandle {
|
||||
this.validateAgentOptions(options.agentOptions ?? {})
|
||||
// Check the agent id BEFORE preparing the session: register() would reject a
|
||||
// duplicate id only AFTER the session enters the store, leaving an orphaned
|
||||
// live session (and lazy persistence state) that blocks reuse of that id.
|
||||
@@ -196,6 +222,7 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
* @returns the handle for the agent resumed on the reconstructed session.
|
||||
*/
|
||||
async resume(options: ResumeAgentOptions): Promise<AgentHandle> {
|
||||
this.validateAgentOptions(options.agentOptions ?? {})
|
||||
// Read the service through `ctx.get('sessionPersistence')` — a direct
|
||||
// global-store lookup keyed by the isolate symbol — NOT
|
||||
// `this.ctx.sessionPersistence`. AgentLoop deliberately does NOT inject
|
||||
@@ -228,6 +255,7 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
* AgentLoop's static inject, so they resolve fine).
|
||||
*/
|
||||
private async resumeWith(persistence: SessionPersistence, options: ResumeAgentOptions): Promise<AgentHandle> {
|
||||
this.validateAgentOptions(options.agentOptions ?? {})
|
||||
this.assertAgentIdFree(options.agentId)
|
||||
const { meta, events } = await persistence.load(options.resumeSessionId)
|
||||
// Re-check the agent id AFTER the await: the pre-load check above can go
|
||||
@@ -267,6 +295,14 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
}
|
||||
}
|
||||
|
||||
/** Validate merge-extended options the loop owns before any session is prepared or loaded. */
|
||||
private validateAgentOptions(options: AgentOptions): void {
|
||||
const { maxParallelToolCalls } = options
|
||||
if (maxParallelToolCalls !== undefined && (!Number.isInteger(maxParallelToolCalls) || maxParallelToolCalls < 1)) {
|
||||
throw new Error('maxParallelToolCalls must be a positive integer')
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Shared: construct a ReactLoopAgent over a PREPARED (not-yet-entered)
|
||||
* session, then build the ONE composite effect that owns the whole agent
|
||||
|
||||
@@ -10,7 +10,7 @@
|
||||
import type { Context } from 'cordis'
|
||||
import type { FinishReason, GenerateOptions, LlmCallConfig, Message } from '@deepseek-ai/dsh-llm'
|
||||
import { BlockAssembler, HarnessError, deepFreeze } from '@deepseek-ai/dsh-llm'
|
||||
import type { ContinuationDecision, HookContext, PromptDecision } from '@deepseek-ai/dsh-agent'
|
||||
import type { ContinuationDecision, PromptDecision } from '@deepseek-ai/dsh-agent'
|
||||
import { canonicalHeader } from '@deepseek-ai/dsh-session'
|
||||
import type { Session, TurnEndReason, TurnTrigger } from '@deepseek-ai/dsh-session'
|
||||
import { createTransmissionLog, recordRequestHeader } from './request-log.ts'
|
||||
@@ -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'
|
||||
|
||||
/** An Error with an optional machine-readable code (e.g., from LlmError or a throwing plugin). */
|
||||
@@ -172,11 +173,12 @@ export interface LoopHandle {
|
||||
* session('assistant/chunk')
|
||||
* msg = waterfall agent/step-result ⟵ BEFORE the log append, so the
|
||||
* session('assistant/message' {content, usage?}) session records what actually ran
|
||||
* each tool-call in msg (sequential, abort-checked):
|
||||
* session('tool/call'); ctx.tools.execute() ⟵ tools/pre-execute (allow/deny/ask)
|
||||
* → dispatch → tools/post-execute
|
||||
* session('tool/result')
|
||||
* append buffered post-execute additionalContext → session('context/message')(s)
|
||||
* schedule tool-calls in msg by ctx.tools.executionMode (exclusive = barrier;
|
||||
* consecutive parallel-safe = one rolling-pool group, ≤ maxParallelToolCalls in flight):
|
||||
* each STARTED call: session('tool/call'); tools/pre-execute (MODEL order)
|
||||
* → tools/execute dispatch/body (parallel pool) → tools/post-execute (MODEL order)
|
||||
* session('tool/result') committed in MODEL order (slot-buffered)
|
||||
* append buffered post-execute additionalContext (model order) → session('context/message')(s)
|
||||
* drain steering → session('steering/message')
|
||||
* session('step/end') ⟵ durable step boundary (no agent/* mirror)
|
||||
* cont = waterfall agent/turn-continuation ⟵ ContinuationDecision; default
|
||||
@@ -864,68 +866,26 @@ async function runStep(
|
||||
)
|
||||
}
|
||||
|
||||
// --- Tool execution (sequential; parallel execution is a TODO) ---
|
||||
// ToolRegistry.execute converts tool failures (including aborts) into
|
||||
// isError results, so abort is re-checked around every call here.
|
||||
// --- Tool execution (scheduled by per-call concurrency safety) ---
|
||||
// executeToolCalls groups the step's calls by ctx.tools.executionMode and runs
|
||||
// parallel-safe runs through a rolling pool. Only dispatch/body overlaps:
|
||||
// tools/pre-execute and tools/post-execute run in model order, tool/result is
|
||||
// committed in model order, and the returned additionalContext buffer is
|
||||
// ordered the same way. Tool failures (including aborts) become isError
|
||||
// results; the scheduler re-checks the shared signal around calls and throws
|
||||
// the abort so this step's caller ends the turn.
|
||||
const toolCalls = message.content.filter(block => block.type === 'tool-call')
|
||||
// Per-step buffer of `additionalContext` attached by tools/post-execute
|
||||
// listeners. Appended as context/message(s) only AFTER every tool/result for
|
||||
// the step, so a multi-call step keeps tool-call/result adjacency
|
||||
// (interleaving context between a call's result and the next call's would
|
||||
// break the pairing the next model request relies on).
|
||||
const pendingContext: HookContext[] = []
|
||||
for (const call of toolCalls) {
|
||||
/* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
|
||||
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
|
||||
const callEvent = session.append('tool/call', { turn, step, callId: call.id, name: call.name, arguments: call.arguments })
|
||||
let parsedArguments: unknown
|
||||
try {
|
||||
parsedArguments = call.arguments ? JSON.parse(call.arguments) : {}
|
||||
} catch {
|
||||
parsedArguments = call.arguments
|
||||
}
|
||||
// TODO(pre-tool-input-rewrite): tools/pre-execute deliberately cannot rewrite
|
||||
// `arguments` — tool/call (the audit record) and assistant/message (the
|
||||
// model-history source) are logged BEFORE execute, and live consumers (ACP,
|
||||
// tool-bash presentation) read the pre-execution args, so an execution-only
|
||||
// rewrite would desync the UI from what ran. Designing that consistently is
|
||||
// its own proposed RFC (docs/rfc/proposed/feature/…-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,
|
||||
// The correlation id MUST be the loop's authoritative call.id (the
|
||||
// model-transcript id that deriveMessages turns into toolCallId), NOT
|
||||
// result.callId — a post-execute waterfall listener returning a
|
||||
// mismatched id would otherwise orphan the call↔result pairing in the
|
||||
// next model request. A listener-internal id, if ever needed, belongs in
|
||||
// a separate diagnostic field, never overloaded onto callId.
|
||||
callId: call.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: [callEvent.seq] })
|
||||
// Buffer (don't append yet) any post-execute additionalContext for this call.
|
||||
if (result.additionalContext) pendingContext.push(result.additionalContext)
|
||||
// signal CAN flip during the await above (abort() inside a tool);
|
||||
// the analyzer can't see through the await boundary.
|
||||
/* v8 ignore start -- signal.reason default unreachable: cancel()/disposal always set it */
|
||||
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
|
||||
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
|
||||
/* v8 ignore stop */
|
||||
}
|
||||
const pendingContext = await executeToolCalls(ctx, agent, turn, step, toolCalls, signal)
|
||||
|
||||
// Append buffered post-execute context AFTER every tool/result, preserving
|
||||
// tool-call/result adjacency across the whole batch. inject() appends into the
|
||||
// open turn (a context/message at its chronological position).
|
||||
// open turn (a context/message at its chronological position). The scheduler
|
||||
// returns the buffer in model call order.
|
||||
for (const context of pendingContext) {
|
||||
agent.inject(context.content, { source: context.source })
|
||||
}
|
||||
|
||||
326
packages/core/agent-loop/src/tool-calls.ts
Normal file
326
packages/core/agent-loop/src/tool-calls.ts
Normal file
@@ -0,0 +1,326 @@
|
||||
/**
|
||||
* The agent loop's per-step tool-call scheduler. `runStep` (loop.ts) hands it
|
||||
* the assistant message's `tool-call` blocks; this module parses each call's
|
||||
* arguments once, classifies it via `ctx.tools.executionMode`, partitions the
|
||||
* calls into ordered groups (one exclusive call, or a run of consecutive
|
||||
* parallel-safe calls), and executes each group — a parallel group through a
|
||||
* rolling pool bounded by the agent's `maxParallelToolCalls`.
|
||||
*
|
||||
* The session log stays the source of truth and is reconstructable regardless
|
||||
* of dispatch timing: each STARTED call appends its own `tool/call` before its
|
||||
* body runs, `tool/result` events are appended in MODEL order (slot-buffered
|
||||
* behind a commit cursor), and buffered `additionalContext` is injected in model
|
||||
* call order after every result. A `tool/call`'s log position may interleave
|
||||
* with a sibling's `tool/result` as the pool replenishes; that is safe because
|
||||
* `tool/call` is log-only and derived history pairs the assistant message's
|
||||
* `tool-call` blocks with the ordered `tool/result`s by `callId`.
|
||||
*
|
||||
* @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 ToolExecution, type ToolExecutionResult } from '@deepseek-ai/dsh-tools'
|
||||
import type { ReactLoopAgent } from './agent.ts'
|
||||
import { DEFAULT_MAX_PARALLEL_TOOL_CALLS } from './constants.ts'
|
||||
|
||||
/** One tool call after argument parsing, ready to schedule. */
|
||||
interface PlannedCall {
|
||||
/** The model-transcript call (authoritative `id`/`name`/raw `arguments`). */
|
||||
block: ToolCallBlock
|
||||
/** The distinct per-call execution object handed to the tool pipeline. */
|
||||
exec: ToolExecution
|
||||
}
|
||||
|
||||
/** A settled call's slot, filled in model order before ordered finalization. */
|
||||
interface Slot {
|
||||
/** The raw dispatch/pre result. */
|
||||
result: ToolExecutionResult
|
||||
/** Whether the result still needs ordered `tools/post-execute` finalization. */
|
||||
needsPost: boolean
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute one assistant step's tool calls, honoring per-call concurrency safety.
|
||||
*
|
||||
* Appends `tool/call` (per started call) and `tool/result` (in model order) to
|
||||
* the session, and returns the ordered `additionalContext` buffer for the loop
|
||||
* to inject after the batch. On abort it drains only already-started calls to
|
||||
* results, drops buffered context, and throws the abort error so `runTurn` owns
|
||||
* the turn-end reason.
|
||||
*
|
||||
* @param ctx - the loop context (reaches `ctx.tools`).
|
||||
* @param agent - the agent being driven (owns the session, options, and is
|
||||
* passed to each `ToolExecution`).
|
||||
* @param turn - the current turn number (for the session events).
|
||||
* @param step - the current step number (for the session events).
|
||||
* @param toolCalls - the assistant message's `tool-call` blocks, in model order.
|
||||
* @param signal - the step's abort signal (shared by every call).
|
||||
* @returns the per-step `additionalContext` buffer in model call order.
|
||||
*/
|
||||
export async function executeToolCalls(
|
||||
ctx: Context,
|
||||
agent: ReactLoopAgent,
|
||||
turn: number,
|
||||
step: number,
|
||||
toolCalls: ToolCallBlock[],
|
||||
signal: AbortSignal,
|
||||
): Promise<HookContext[]> {
|
||||
const { session, options } = agent
|
||||
const maxParallel = options.maxParallelToolCalls ?? DEFAULT_MAX_PARALLEL_TOOL_CALLS
|
||||
|
||||
// Plan: parse each call's raw JSON arguments exactly once, and build one
|
||||
// distinct ToolExecution per call so a `tools/execute` wrapper that mutates
|
||||
// `exec` in place (e.g. replacing exec.signal with a per-call deadline) cannot
|
||||
// race through a shared payload.
|
||||
const planned: PlannedCall[] = toolCalls.map(block => ({
|
||||
block,
|
||||
exec: {
|
||||
callId: block.id,
|
||||
name: block.name,
|
||||
arguments: parseArguments(block.arguments),
|
||||
agent,
|
||||
signal,
|
||||
},
|
||||
}))
|
||||
|
||||
// Partition into ordered groups: an exclusive call is its own group (a
|
||||
// barrier), a run of consecutive parallel-safe calls is one group. Grouping
|
||||
// uses executionMode so an exclusive tool between two reads splits them into
|
||||
// separate ordered groups (no read/write race inside one assistant step).
|
||||
const groups = groupByMode(ctx, planned)
|
||||
|
||||
const pendingContext: HookContext[] = []
|
||||
for (const group of groups) {
|
||||
// Groups are never empty (groupByMode only pushes non-empty runs/singletons).
|
||||
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- non-empty group
|
||||
const first = group[0]!
|
||||
if (group.length === 1 && ctx.tools.executionMode(first.exec).kind === 'exclusive') {
|
||||
await runExclusive(ctx, session, turn, step, first, signal, pendingContext)
|
||||
} else {
|
||||
await runParallelGroup(ctx, session, turn, step, group, signal, maxParallel, pendingContext)
|
||||
}
|
||||
}
|
||||
return pendingContext
|
||||
}
|
||||
|
||||
/** Parse a model-produced raw arguments string, falling back to the raw string on invalid JSON (empty ⇒ `{}`). */
|
||||
function parseArguments(raw: string): unknown {
|
||||
try {
|
||||
return raw ? JSON.parse(raw) : {}
|
||||
} catch {
|
||||
return raw
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Group planned calls into ordered runs: each exclusive call is a singleton
|
||||
* group; consecutive parallel-safe calls coalesce into one group. `executionMode`
|
||||
* is queried once per call here and again by the caller to pick the exclusive
|
||||
* fast-path — both reads are pure and cheap.
|
||||
*/
|
||||
function groupByMode(ctx: Context, planned: PlannedCall[]): PlannedCall[][] {
|
||||
const groups: PlannedCall[][] = []
|
||||
let run: PlannedCall[] = []
|
||||
const flush = (): void => {
|
||||
if (run.length > 0) {
|
||||
groups.push(run)
|
||||
run = []
|
||||
}
|
||||
}
|
||||
for (const call of planned) {
|
||||
if (ctx.tools.executionMode(call.exec).kind === 'parallel') {
|
||||
run.push(call)
|
||||
} else {
|
||||
flush()
|
||||
groups.push([call])
|
||||
}
|
||||
}
|
||||
flush()
|
||||
return groups
|
||||
}
|
||||
|
||||
/**
|
||||
* The exclusive single-call path keeps the public one-call pipeline sequential:
|
||||
* abort-check, `tool/call`, pre/dispatch/post via `ctx.tools.execute`,
|
||||
* `tool/result`, buffer context, post-await abort-check.
|
||||
*/
|
||||
async function runExclusive(
|
||||
ctx: Context,
|
||||
session: Session,
|
||||
turn: number,
|
||||
step: number,
|
||||
call: PlannedCall,
|
||||
signal: AbortSignal,
|
||||
pendingContext: HookContext[],
|
||||
): Promise<void> {
|
||||
/* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
|
||||
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
|
||||
const callSeq = appendToolCall(session, turn, step, call.block)
|
||||
const result = await ctx.tools.execute(call.exec)
|
||||
appendToolResult(session, turn, step, call.block, result, callSeq)
|
||||
if (result.additionalContext) pendingContext.push(result.additionalContext)
|
||||
// signal CAN flip during the await above (abort() inside a tool); the analyzer
|
||||
// can't see through the await boundary.
|
||||
/* v8 ignore start -- signal.reason default unreachable: cancel()/disposal always set it */
|
||||
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
|
||||
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
|
||||
/* v8 ignore stop */
|
||||
}
|
||||
|
||||
/**
|
||||
* The rolling-pool path for a group of parallel-safe calls. Starts calls in
|
||||
* model order up to `maxParallel`, and whenever one settles starts the next
|
||||
* unstarted call until the group is exhausted. Settled dispatches land in
|
||||
* model-order slots; a commit cursor appends `tool/result` (and collects
|
||||
* `additionalContext`) only while the next slot is ready, so the log stays
|
||||
* model-ordered regardless of completion order.
|
||||
*
|
||||
* Abort: an already-aborted signal starts nothing and throws before any
|
||||
* `tool/call`. An abort mid-group stops replenishment, awaits only the started
|
||||
* calls, commits their results in order, drops buffered context, and throws.
|
||||
*/
|
||||
async function runParallelGroup(
|
||||
ctx: Context,
|
||||
session: Session,
|
||||
turn: number,
|
||||
step: number,
|
||||
group: PlannedCall[],
|
||||
signal: AbortSignal,
|
||||
maxParallel: number,
|
||||
pendingContext: HookContext[],
|
||||
): Promise<void> {
|
||||
/* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
|
||||
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
|
||||
|
||||
const slots: (Slot | undefined)[] = group.map(() => undefined)
|
||||
// callSeqs[i] is the `tool/call` event seq for started slot i (its provenance
|
||||
// for the matching tool/result). A slot is only committed after it is started,
|
||||
// so its callSeq is always set by then.
|
||||
const callSeqs: number[] = group.map(() => -1)
|
||||
let nextToStart = 0
|
||||
let committed = 0
|
||||
let started = 0
|
||||
let aborted: boolean = signal.aborted
|
||||
|
||||
// Advance the commit cursor over contiguous settled slots: run post-execute in
|
||||
// model order, append each tool/result, and collect its additionalContext.
|
||||
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
|
||||
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- bounded index
|
||||
? await ctx.tools[TOOL_REGISTRY_SCHEDULER].finalize(call!.exec, slot.result)
|
||||
: slot.result
|
||||
// committed < group.length, so call and its callSeq (set at start) exist.
|
||||
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- bounded index
|
||||
appendToolResult(session, turn, step, call!.block, result, callSeqs[committed]!)
|
||||
if (result.additionalContext) pendingContext.push(result.additionalContext)
|
||||
committed++
|
||||
}
|
||||
}
|
||||
|
||||
const inFlight = new Map<number, Promise<number>>()
|
||||
|
||||
const startCall = async (index: number): Promise<void> => {
|
||||
// index is always < group.length (bounded by every caller).
|
||||
// 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(call.exec).then((result) => {
|
||||
slots[index] = { result, needsPost: true }
|
||||
return index
|
||||
})
|
||||
inFlight.set(index, promise)
|
||||
break
|
||||
}
|
||||
case 'post-result':
|
||||
slots[index] = { result: prepared.result, needsPost: true }
|
||||
break
|
||||
case 'final-result':
|
||||
slots[index] = { 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) {
|
||||
await startCall(nextToStart)
|
||||
nextToStart++
|
||||
await commitReady()
|
||||
// The signal CAN flip while an ordered pre-execute listener is running.
|
||||
if (signal.aborted) aborted = true
|
||||
}
|
||||
}
|
||||
|
||||
// Prime the pool up to the cap. Ordered pre-execute listeners may be async;
|
||||
// dispatch/body is the only stage that overlaps across in-flight calls.
|
||||
await fillPool()
|
||||
while (inFlight.size > 0) {
|
||||
const settledIndex = await Promise.race(inFlight.values())
|
||||
inFlight.delete(settledIndex)
|
||||
// Commit every contiguous settled slot now available.
|
||||
await commitReady()
|
||||
// The signal CAN flip during the await above (abort() inside a tool); the
|
||||
// analyzer can't see through the await boundary. An abort stops the pool
|
||||
// from starting any further calls, but already-started calls still drain.
|
||||
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
|
||||
if (signal.aborted) aborted = true
|
||||
await fillPool()
|
||||
}
|
||||
|
||||
if (aborted) {
|
||||
// Every started call has settled and committed in order; buffered context
|
||||
// from this aborted step is dropped (not injected). Raise the abort so the
|
||||
// existing runTurn catch owns turn/end reason selection. Unstarted calls
|
||||
// beyond the cap never appended a tool/call.
|
||||
/* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
|
||||
throw new Error(String(signal.reason ?? 'aborted'))
|
||||
}
|
||||
// A defensive check the started count matches what we committed — a parallel
|
||||
// group with no abort commits every started slot, and started === group.length.
|
||||
/* v8 ignore next -- unreachable: a non-aborted group starts and commits all calls */
|
||||
if (committed !== started) throw new Error('tool-call scheduler: uncommitted settled calls')
|
||||
}
|
||||
|
||||
/** Append the `tool/call` audit event for one started call; returns its seq (the tool/result's provenance). */
|
||||
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 one call's `tool/result`, keyed by the authoritative model-transcript call id and provenanced to its `tool/call`. */
|
||||
function appendToolResult(
|
||||
session: Session,
|
||||
turn: number,
|
||||
step: number,
|
||||
block: ToolCallBlock,
|
||||
result: ToolExecutionResult,
|
||||
callSeq: number,
|
||||
): void {
|
||||
session.append('tool/result', {
|
||||
turn, step,
|
||||
// The correlation id MUST be the loop's authoritative call.id (the
|
||||
// model-transcript id deriveMessages turns into toolCallId), NOT
|
||||
// result.callId — a post-execute listener returning a mismatched id would
|
||||
// otherwise orphan the call↔result pairing in the next model request.
|
||||
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] })
|
||||
}
|
||||
462
packages/core/agent-loop/tests/tool-calls.spec.ts
Normal file
462
packages/core/agent-loop/tests/tool-calls.spec.ts
Normal file
@@ -0,0 +1,462 @@
|
||||
/**
|
||||
* The per-step tool-call scheduler (`tool-calls.ts`): grouping by
|
||||
* `ctx.tools.executionMode`, the rolling pool for parallel groups, model-order
|
||||
* `tool/result` commit despite out-of-order settlement, interleaved `tool/call`
|
||||
* audit records, ordered `tools/pre-execute`/`tools/post-execute`,
|
||||
* model-ordered `additionalContext`, and abort behavior.
|
||||
*
|
||||
* Tools are mocked and deterministic — no real API, no snapshot here (the
|
||||
* transcript-facing live-order behavior is pinned by the ACP snapshot goldens).
|
||||
*/
|
||||
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import { CallId, StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
import SessionStore, { SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
|
||||
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
|
||||
import LlmService from '@deepseek-ai/dsh-llm'
|
||||
import ToolRegistry, { defineTool, type PostToolDecision, type PreToolDecision } from '@deepseek-ai/dsh-tools'
|
||||
import AgentRegistry, { AgentId } from '@deepseek-ai/dsh-agent'
|
||||
import AgentLoop, { ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
|
||||
import { MockAdapter, textResponse } from './mock-adapter.ts'
|
||||
|
||||
async function harness(adapter: MockAdapter) {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(SystemPrompt, { persona: '' })
|
||||
await ctx.plugin(ToolRegistry)
|
||||
await ctx.plugin(AgentRegistry)
|
||||
await ctx.plugin(AgentLoop, { agents: [] })
|
||||
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]
|
||||
}
|
||||
|
||||
/** An assistant message with N tool-call blocks named `name` (ids c1..cN, arg = index). */
|
||||
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 parallel-safe tool whose calls block until the test releases them by callId. */
|
||||
function gatedParallelTool(name: string) {
|
||||
const gates = new Map<string, () => void>()
|
||||
const started: string[] = []
|
||||
const tool = defineTool({
|
||||
name,
|
||||
description: `gated ${name}`,
|
||||
parameters: { id: { type: 'string', required: true } },
|
||||
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 one in-flight call by its arg id (its `execute` resolves). */
|
||||
release(id: string) { gates.get(id)?.(); gates.delete(id) },
|
||||
pending() { return [...gates.keys()] },
|
||||
}
|
||||
}
|
||||
|
||||
/** 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'), { model: 'mock' })
|
||||
|
||||
agent.send([{ type: 'text', text: 'go' }])
|
||||
// All three start before any is released — proof of concurrency.
|
||||
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 () => {
|
||||
// read A (safe), write A (exclusive), read A (safe) → the write must not
|
||||
// overlap either read. The exclusive tool records whether a read was still
|
||||
// in flight when it ran.
|
||||
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'), { model: 'mock' })
|
||||
agent.send([{ type: 'text', text: 'go' }])
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
// The write ran strictly between the two reads (barrier ordering).
|
||||
expect(order).toEqual(['r-start-A1', 'r-end-A1', 'w-A2', 'r-start-A3', 'r-end-A3'])
|
||||
})
|
||||
})
|
||||
|
||||
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'), { model: 'mock' })
|
||||
|
||||
agent.send([{ type: 'text', text: 'go' }])
|
||||
await until(() => gated.started.length === 2)
|
||||
// Release the SECOND call first; its result must NOT be committed until the
|
||||
// first commits (the commit cursor holds it in a slot).
|
||||
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'), { model: 'mock' })
|
||||
agent.send([{ type: 'text', text: 'go' }])
|
||||
await until(() => gated.started.length === 2)
|
||||
gated.release('2'); gated.release('1')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
// deriveMessages pairs the assistant tool-call blocks with tool-result
|
||||
// blocks by callId — model order, independent of log interleaving.
|
||||
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 programmatic maxParallelToolCalls values before creating agents', async () => {
|
||||
const ctx = await harness(new MockAdapter([]))
|
||||
|
||||
expect(() => ctx.agentLoop.create(AgentId('bad-zero'), { model: 'mock', maxParallelToolCalls: 0 }))
|
||||
.toThrow('maxParallelToolCalls must be a positive integer')
|
||||
expect(() => ctx.agentLoop.createAgent({
|
||||
agentId: AgentId('bad-fractional'),
|
||||
sessionId: SessionId('bad-fractional-session'),
|
||||
agentOptions: { model: 'mock', maxParallelToolCalls: 1.5 },
|
||||
})).toThrow('maxParallelToolCalls must be a positive integer')
|
||||
})
|
||||
|
||||
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)
|
||||
const gated = gatedParallelTool('p')
|
||||
ctx.tools.register(gated.tool)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock', maxParallelToolCalls: 2 })
|
||||
|
||||
agent.send([{ type: 'text', text: 'go' }])
|
||||
// Only 2 start initially (the cap).
|
||||
await until(() => gated.started.length === 2)
|
||||
await new Promise(r => setTimeout(r, 5))
|
||||
expect(gated.started).toEqual(['1', '2'])
|
||||
// Releasing one starts the next in model order.
|
||||
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)
|
||||
const gated = gatedParallelTool('p')
|
||||
ctx.tools.register(gated.tool)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock', maxParallelToolCalls: 1 })
|
||||
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 additionalContext', () => {
|
||||
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'), { model: 'mock' })
|
||||
|
||||
agent.send([{ type: 'text', text: 'go' }])
|
||||
await until(() => gated.started.length === 3)
|
||||
// Settle in reverse; post-execute (ordered by the commit cursor) still fires
|
||||
// in model order because post runs on the commit path, not on dispatch.
|
||||
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 additionalContext 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)
|
||||
const gated = gatedParallelTool('p')
|
||||
ctx.tools.register(gated.tool)
|
||||
ctx.on('tools/post-execute', async (exec, _result): Promise<PostToolDecision> =>
|
||||
({ kind: 'accept', additionalContext: { content: [{ type: 'text', text: `ctx-${exec.callId}` }], source: { kind: 'plugin', plugin: 'p' } } }))
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { 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)
|
||||
// Both tool/results precede both context/messages, and context is model-ordered.
|
||||
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('keeps pre-produced deny/error results ordered without dispatching those calls', 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'), { 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'), { model: 'mock' })
|
||||
ctx.on('session/event', (session, event) => {
|
||||
if (session === agent.session && event.type === 'assistant/message') {
|
||||
;(agent as unknown as { currentAbort?: AbortController }).currentAbort?.abort('already aborted')
|
||||
}
|
||||
})
|
||||
|
||||
agent.send([{ type: 'text', text: 'go' }])
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(gated.started).toEqual([])
|
||||
expect(events(agent).filter(e => e.type === 'tool/call')).toEqual([])
|
||||
expect(events(agent).filter(e => e.type === 'tool/result')).toEqual([])
|
||||
})
|
||||
|
||||
it('stops starting siblings when abort fires during ordered pre-execute', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
multiCall([{ id: 'c1', name: 'p', args: { id: '1' } }, { id: 'c2', name: 'p', args: { id: '2' } }]),
|
||||
textResponse('should never be requested'),
|
||||
])
|
||||
const ctx = await harness(adapter)
|
||||
const gated = gatedParallelTool('p')
|
||||
ctx.tools.register(gated.tool)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
ctx.on('tools/pre-execute', async (exec, next): Promise<PreToolDecision> => {
|
||||
if (exec.callId === CallId('c1')) {
|
||||
;(agent as unknown as { currentAbort?: AbortController }).currentAbort?.abort('pre cancelled')
|
||||
}
|
||||
return next()
|
||||
})
|
||||
|
||||
agent.send([{ type: 'text', text: 'go' }])
|
||||
await until(() => gated.started.length === 1)
|
||||
await new Promise(r => setTimeout(r, 5))
|
||||
expect(gated.started).toEqual(['1'])
|
||||
gated.release('1')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(events(agent).filter(e => e.type === 'tool/call').map(e => e.data.callId))
|
||||
.toEqual([CallId('c1')])
|
||||
expect(events(agent).filter(e => e.type === 'tool/result').map(e => e.data.callId))
|
||||
.toEqual([CallId('c1')])
|
||||
})
|
||||
|
||||
it('stops replenishing after abort, commits started results, and drops buffered additionalContext', 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)
|
||||
const gated = gatedParallelTool('p')
|
||||
ctx.tools.register(gated.tool)
|
||||
ctx.on('tools/post-execute', async (exec, _result, next): Promise<PostToolDecision> => ({
|
||||
...await next(),
|
||||
additionalContext: { content: [{ type: 'text', text: `ctx-${exec.callId}` }], source: { kind: 'plugin', plugin: 'p' } },
|
||||
}))
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock', maxParallelToolCalls: 2 })
|
||||
|
||||
agent.send([{ type: 'text', text: 'go' }])
|
||||
await until(() => gated.started.length === 2)
|
||||
;(agent as unknown as { currentAbort?: AbortController }).currentAbort?.abort('stop now')
|
||||
gated.release('1')
|
||||
gated.release('2')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(gated.started).toEqual(['1', '2'])
|
||||
expect(events(agent).filter(e => e.type === 'tool/call').map(e => e.data.callId))
|
||||
.toEqual([CallId('c1'), CallId('c2')])
|
||||
expect(events(agent).filter(e => e.type === 'tool/result').map(e => e.data.callId))
|
||||
.toEqual([CallId('c1'), CallId('c2')])
|
||||
expect(events(agent).filter(e => e.type === 'context/message')).toEqual([])
|
||||
})
|
||||
|
||||
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)
|
||||
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'), { model: 'mock', maxParallelToolCalls: 2 })
|
||||
|
||||
agent.send([{ type: 'text', text: 'go' }])
|
||||
await until(() => gated.started.length === 2)
|
||||
;(agent as unknown as { currentAbort?: AbortController }).currentAbort?.abort('stop before barrier')
|
||||
gated.release('1')
|
||||
gated.release('2')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(exclusive).toEqual([])
|
||||
expect(events(agent).filter(e => e.type === 'tool/call').map(e => e.data.callId))
|
||||
.toEqual([CallId('c1'), CallId('c2')])
|
||||
})
|
||||
})
|
||||
Reference in New Issue
Block a user