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

docs: tighten parallel tool-call prose

Browse Source
This commit is contained in:
Tianyi Cui
2026-07-18 14:59:26 +08:00
parent 3e07f9b270
commit 21ec178841
30 changed files with 161 additions and 320 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/core/agent-loop/src/agent.ts
+2 -2
View File
@@ -54,7 +54,7 @@ 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 scheduler cap shared by this factory's agents.
* @param maxParallelToolCalls - resolved in-flight cap for this agent.
* @returns the agent and closures bound only to that exact instance.
*/
export function prepareReactLoopAgent(
@@ -144,7 +144,7 @@ export class ReactLoopAgent implements Agent {
* the `disposed` transition fires and leave the promise hanging.
*/
private idleWaiters: (() => void)[] = []
/** Immutable scheduler cap resolved by the owning AgentLoop factory. */
/** Maximum parallel-safe calls allowed in one step. */
private readonly maxParallelToolCalls: number
/**
* Durability checkpoints started by idle {@link inject} calls. `inject()` is
packages/core/agent-loop/src/constants.ts
+2 -11
View File
@@ -1,15 +1,6 @@
/**
* 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.
*
/** Shared agent-loop scheduler defaults.
* @module dsh-agent-loop/constants
*/
/**
* Default cap on simultaneously in-flight tool calls within one assistant step
* when the agent-loop config omits one. Matches the rolling-pool size Claude
* Code uses; a larger group is not truncated — the cap limits concurrency, not
* the group.
*/
/** Default maximum in-flight parallel-safe calls per agent step. */
export const DEFAULT_MAX_PARALLEL_TOOL_CALLS = 10
packages/core/agent-loop/src/index.ts
+3 -5
View File
@@ -333,9 +333,8 @@ export { DEFAULT_MAX_PARALLEL_TOOL_CALLS }
/** Agent-loop plugin configuration. */
export interface Config {
/**
* Concurrent parallel-safe tool-call cap shared by every agent this factory
* creates. A positive integer; `1` preserves fully serial execution and an
* omitted value defaults to {@link DEFAULT_MAX_PARALLEL_TOOL_CALLS}.
* 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. */
@@ -355,7 +354,6 @@ export class AgentLoop extends Service implements AgentFactory {
/** Runtime schema for declarative agents. */
static Config = z.object({
// The deployment-wide cap is defaulted and validated at plugin load.
maxParallelToolCalls: z.number().step(1).min(1).default(DEFAULT_MAX_PARALLEL_TOOL_CALLS),
agents: z.array(z.object({
id: z.string().required(),
@@ -367,7 +365,7 @@ export class AgentLoop extends Service implements AgentFactory {
}) as unknown as z<Config>
private readonly ownership: FactoryOwnership
/** Resolved immutable scheduler cap shared by every driver from this factory. */
/** 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 }
packages/core/agent-loop/src/loop.ts
+3 -4
View File
@@ -74,7 +74,7 @@ function stepFinishReason(finish: FinishReason): TurnEndReason | undefined {
export interface LoopHandle {
/** Native-private agent inbox handed to the driver only at internal startup. */
readonly inbox: Inbox
/** Immutable concurrent tool-call cap resolved by the owning factory. */
/** Maximum parallel-safe calls allowed in one step. */
readonly maxParallelToolCalls: number
setStatus(status: 'idle' | 'running'): void
setAbort(controller: AbortController | undefined): void
@@ -558,13 +558,12 @@ async function runStep(
// Empty messages exist only to carry usage; the helper also omits empty chunk provenance.
recordAssistantMessage(session, turn, step, header.config, assembledContent, message, assembler, chunkSeqs)
// The scheduler overlaps only dispatch/body for parallel-safe calls; policy,
// results, and additional context remain in model order.
// Dispatch may overlap; policy, results, and context remain model-ordered.
const pendingContext = toolCalls.length > 0
? await executeToolCalls(ctx, agent, turn, step, toolCalls, signal, maxParallelToolCalls)
: []
// Append context after the complete result batch to preserve call/result adjacency.
// Context follows the complete result batch to preserve call/result adjacency.
for (const context of pendingContext) {
agent.inject(context.content, {
source: context.source,
packages/core/agent-loop/src/tool-calls.ts
+35 -84
View File
@@ -1,22 +1,11 @@
/**
* 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 pending calls via `ctx.tools.executionMode`, and
* runs ordered groups through a rolling pool bounded by the agent-loop's
* `maxParallelToolCalls` config. Exclusive calls are singleton barriers. A
* parallel group reclassifies each later call before it starts, so registry
* changes during an earlier barrier or ordered result commit take effect before
* the pool replenishes.
*
* 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 `additionalContexts` are 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`.
* 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 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
*/
@@ -29,40 +18,30 @@ import type { ReactLoopAgent } from './agent.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 input handed to the tool pipeline. */
exec: ToolExecutionInput
}
/** A settled call's slot, filled in model order before ordered finalization. */
/** Settled dispatch awaiting model-order finalization. */
interface Slot {
/** The registry-minted execution object, carrying this call's token. */
exec: ToolRunContext
/** 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.
* Schedule one assistant step's tool calls by their live concurrency mode.
* Started calls receive ordered results; abort drains them, discards their
* buffered context, and rethrows so the turn owns final error handling.
*
* Appends `tool/call` (per started call) and `tool/result` (in model order) to
* the session, and returns the ordered `additionalContexts` 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).
* @param maxParallel - the already-validated cap snapshot for parallel groups.
* @returns the per-step `additionalContexts` buffer in model call order.
* @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.
* @returns buffered contexts in model call order.
*/
export async function executeToolCalls(
ctx: Context,
@@ -75,10 +54,7 @@ export async function executeToolCalls(
): Promise<HookContext[]> {
const { session } = agent
// 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.
// Inputs are distinct because tools/execute wrappers may replace `exec.signal`.
const planned: PlannedCall[] = toolCalls.map(block => ({
block,
exec: {
@@ -93,9 +69,7 @@ export async function executeToolCalls(
const pendingContext: HookContext[] = []
let next = 0
while (next < planned.length) {
// Classify the next group only after the previous one has fully committed.
// A registry mutation in an exclusive call or result observer therefore
// changes how every not-yet-started call is scheduled.
// 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
@@ -105,7 +79,7 @@ export async function executeToolCalls(
return pendingContext
}
/** Parse a model-produced raw arguments string, falling back to the raw string on invalid JSON (empty ⇒ `{}`). */
/** Parse model arguments, preserving invalid JSON as text and mapping empty input to `{}`. */
function parseArguments(raw: string): unknown {
try {
return raw ? JSON.parse(raw) : {}
@@ -115,18 +89,11 @@ function parseArguments(raw: string): unknown {
}
/**
* The rolling-pool path for one ordered group. A singleton exclusive group runs
* as a pool of one (a barrier). A parallel-safe run starts calls in model order
* up to `maxParallel`; before each later call starts, the scheduler reclassifies
* it against the live registry. An exclusive result stops replenishment, drains
* the current run, and remains for the caller's next singleton group. Settled
* dispatches land in model-order slots; a commit cursor appends `tool/result`
* (and collects `additionalContexts`) 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.
* 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, discards
* their contexts, and throws.
*/
async function runGroup(
ctx: Context,
@@ -142,17 +109,14 @@ async function runGroup(
/* 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.
// 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
// Advance the commit cursor over contiguous settled slots: run post-execute in
// model order, append each tool/result, and collect its additionalContexts.
// `committed` advances only across contiguous model-order slots.
const commitReady = async (): Promise<void> => {
while (committed < group.length) {
const slot = slots[committed]
@@ -161,7 +125,6 @@ async function runGroup(
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)
// 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]!)
pendingContext.push(...result.additionalContexts ?? [])
@@ -172,7 +135,6 @@ async function runGroup(
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)
@@ -201,9 +163,7 @@ async function runGroup(
const fillPool = async (): Promise<void> => {
while (!aborted && nextToStart < group.length && inFlight.size < maxParallel) {
// The caller classified the first item immediately before entering this
// group. Re-read every later item after ordered commits so a live registry
// change can turn it into the next barrier.
// 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'
@@ -211,49 +171,40 @@ async function runGroup(
await startCall(nextToStart)
nextToStart++
await commitReady()
// The signal CAN flip while an ordered pre-execute listener is running.
// Abort may arrive while pre-execute awaits.
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.
// Ordered pre-execute may await; only dispatch/body overlaps.
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.
// Abort may arrive while a tool or ordered commit awaits.
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
if (signal.aborted) aborted = true
await fillPool()
}
if (aborted) {
// 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.
// Started calls are committed; their context is discarded with the aborted step.
/* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
throw new Error(String(signal.reason ?? 'aborted'))
}
// A defensive check that every started call committed before this group
// returns; a reclassified barrier may leave the rest of `group` unstarted.
/* 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 the `tool/call` audit event for one started call; returns its seq (the tool/result's provenance). */
/** 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 one call's `tool/result`, keyed by the authoritative model-transcript call id and provenanced to its `tool/call`. */
/** Append a model-ordered result linked to its call event. */
function appendToolResult(
session: Session,
turn: number,