refactor(agent-loop): unify tool-call scheduling on one rolling pool + factory cap default

Run every ordered group through the same rolling pool: an exclusive call is a
pool of one (a barrier), dropping the separate runExclusive path and the
redundant post-grouping executionMode re-query. Behavior is unchanged — the
parallel-tool-calls snapshot and the full scheduler unit suite (barriers, cap,
abort, model-order results) stay green.

Add AgentLoop.Config.maxParallelToolCalls as a factory-wide default applied to
every agent create/createAgent/resume mints (per-agent option overrides it),
forwarded through agent-core so it reaches front doors that expose no cap field
of their own. Trim the isConcurrencySafe JSDoc to the local contract and link
the parallel-tool-call RFC for the full rationale; document the field on the
canonical core-data-structures page.
This commit is contained in:
Dudu-0223
2026-07-16 11:36:16 +08:00
parent 77be4b891b
commit 3b1d1bfa12
12 changed files with 162 additions and 88 deletions

View File

@@ -344,6 +344,16 @@ export { DEFAULT_MAX_PARALLEL_TOOL_CALLS } from './constants.ts'
/** Plugin configuration for declarative startup agents. */
export interface Config {
/**
* Default concurrent tool-call cap applied to every agent this factory
* creates (declarative startup agents and factory callers such as the ACP,
* stdio, and SDK front doors that go through `create`/`createAgent`/`resume`).
* A positive integer; a per-agent `maxParallelToolCalls` overrides it, and an
* agent with neither falls back to {@link DEFAULT_MAX_PARALLEL_TOOL_CALLS}.
* This is the single `cordis.yml` knob that reaches agents whose front door
* does not expose its own cap field.
*/
maxParallelToolCalls?: number
/** Agents created or resumed at plugin startup. */
agents: (AgentOptions & {
/** Registry identity for the live agent. */
@@ -366,6 +376,9 @@ export class AgentLoop extends Service implements AgentFactory {
/** Runtime schema for declarative agents. */
static Config = z.object({
// The factory-wide default cap; a per-agent value overrides it. A positive
// integer, validated here so a bad cordis.yml value fails at load.
maxParallelToolCalls: z.number().step(1).min(1),
agents: z.array(z.object({
id: z.string().required(),
model: z.string(),
@@ -410,6 +423,22 @@ export class AgentLoop extends Service implements AgentFactory {
}
}
/**
* Merge the factory-wide default cap into one agent's options. A per-agent
* `maxParallelToolCalls` wins; otherwise the `Config.maxParallelToolCalls`
* default applies, reaching factory callers (ACP/stdio/SDK front doors) whose
* own config does not set a cap. Absent both, the loop falls back to
* {@link DEFAULT_MAX_PARALLEL_TOOL_CALLS} at schedule time.
* @param options - the caller-supplied agent options.
* @returns options with the default cap applied when the caller omitted one.
*/
private withFactoryDefaults(options: AgentOptions): AgentOptions {
if (options.maxParallelToolCalls !== undefined || this.config.maxParallelToolCalls === undefined) {
return options
}
return { ...options, maxParallelToolCalls: this.config.maxParallelToolCalls }
}
/**
* Create an agent on a fresh per-run session, owned by the accessing fiber.
* Constructor-driven config calls use the loop fiber itself.
@@ -419,13 +448,14 @@ export class AgentLoop extends Service implements AgentFactory {
* @returns the published running agent.
*/
create(id: AgentId, options: AgentOptions = {}, meta: Pick<SessionHeader, 'cwd'> = {}): ReactLoopAgent {
validateAgentOptions(options)
const resolved = this.withFactoryDefaults(options)
validateAgentOptions(resolved)
const loopCtx = this.runtime.ctx
const transaction = new AgentCreationTransaction(loopCtx, this.ctx, this.ownership, id)
try {
const sessionId = SessionId(`${id}-session-${randomUUID()}`)
const session = loopCtx.sessions.prepare(sessionId, { meta })
const agent = transaction.prepare(options, session)
const agent = transaction.prepare(resolved, session)
transaction.publish('startup')
return agent
} catch (error: unknown) {
@@ -443,7 +473,8 @@ export class AgentLoop extends Service implements AgentFactory {
* @returns the published handle.
*/
async createAgent(ownerCtx: Context, options: CreateAgentOptions): Promise<AgentHandle> {
validateAgentOptions(options.agentOptions ?? {})
const agentOptions = this.withFactoryDefaults(options.agentOptions ?? {})
validateAgentOptions(agentOptions)
const transaction = new AgentCreationTransaction(
this.runtime.ctx,
ownerCtx,
@@ -456,7 +487,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)
await transaction.waitFor(options.setup?.(agent.ctx))
transaction.assertActive()
return transaction.publish('startup')
@@ -475,7 +506,6 @@ export class AgentLoop extends Service implements AgentFactory {
* @returns the published handle.
*/
async resume(ownerCtx: Context, options: ResumeAgentOptions): Promise<AgentHandle> {
validateAgentOptions(options.agentOptions ?? {})
const persistence = this.runtime.ctx.get('sessionPersistence')
if (persistence === undefined) {
throw new Error('cannot resume: session persistence is not configured (load a dsh-session-persistence backend)')
@@ -489,6 +519,8 @@ export class AgentLoop extends Service implements AgentFactory {
persistence: SessionPersistence,
options: ResumeAgentOptions,
): Promise<AgentHandle> {
const agentOptions = this.withFactoryDefaults(options.agentOptions ?? {})
validateAgentOptions(agentOptions)
const transaction = new AgentCreationTransaction(
this.runtime.ctx,
ownerCtx,
@@ -508,7 +540,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)
await transaction.waitFor(options.setup?.(agent.ctx))
transaction.assertActive()
return transaction.publish('resume')

View File

@@ -3,8 +3,9 @@
* 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`.
* parallel-safe calls), and runs every group through the same rolling pool
* bounded by the agent's `maxParallelToolCalls` — an exclusive group is a pool
* of one.
*
* 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
@@ -95,16 +96,13 @@ export async function executeToolCalls(
// separate ordered groups (no read/write race inside one assistant step).
const groups = groupByMode(ctx, planned)
// Every group runs through the same rolling pool: an exclusive call is a
// singleton group (pool of one, a barrier), a parallel-safe run is one group
// bounded by the cap. `groupByMode` already classified each call, so the loop
// does not re-query `executionMode` here.
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)
}
await runGroup(ctx, session, turn, step, group, signal, maxParallel, pendingContext)
}
return pendingContext
}
@@ -135,8 +133,8 @@ function parseArguments(raw: string): unknown {
/**
* 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.
* is the sole classification point — the caller runs every group through the
* rolling pool without re-querying it. The read is pure and cheap.
*/
function groupByMode(ctx: Context, planned: PlannedCall[]): PlannedCall[][] {
const groups: PlannedCall[][] = []
@@ -167,46 +165,19 @@ function assertMaxParallelToolCalls(maxParallel: number): void {
}
/**
* 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.
* 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`, 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(
async function runGroup(
ctx: Context,
session: Session,
turn: number,

View File

@@ -274,6 +274,48 @@ describe('tool-call scheduler: rolling pool honors maxParallelToolCalls', () =>
gated.release('2')
await waitForIdle(ctx, agent)
})
it('applies the factory-wide Config default to agents that set no per-agent cap', 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)
// Factory default of 1 (no per-agent cap set below) must serialize.
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'), { model: 'mock' })
expect(agent.options.maxParallelToolCalls).toBe(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)
})
it('lets a per-agent cap override the factory-wide Config default', 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)
await ctx.plugin(AgentLoop, { agents: [], maxParallelToolCalls: 1 })
ctx.llm.registerAdapter(['mock'], new MockAdapter([]))
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock', maxParallelToolCalls: 4 })
expect(agent.options.maxParallelToolCalls).toBe(4)
})
})
describe('tool-call scheduler: ordered middleware and additionalContext', () => {