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

refactor(agent-loop): simplify parallel tool-call cap config

Browse Source
This commit is contained in:
Dudu-0223
2026-07-16 14:36:16 +08:00
parent 3b1d1bfa12
commit 91da66e715
22 changed files with 182 additions and 284 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -43,7 +43,7 @@ import type { Config } from '@deepseek-ai/dsh-agent-core'
// intersects the owner schemas, so validation and defaulting can never drift from the owners.
```
The bundle FORWARDS each field to the child that owns it: `agents` to `agent-loop` (default `[]`), so each app supplies its own pre-created agents — a stdio app pre-creates a `main`; the ACP app pre-creates none (it creates agents on demand at `session/new`) — `maxParallelToolCalls` to `agent-loop` as the factory-wide default concurrent tool-call cap for every agent it creates; `persona` and `toolOrder` to `dsh-system-prompt`; `tools` to the tool registry for its presentation mode; and `skills.registry`, `skills.local`, and `skills.tool` to the skill registry, local provider, and model-facing consumer. Forwarding is exactly why the owners can live in the shared spine even though the apps disagree on what to configure.
The bundle FORWARDS each field to the child that owns it: `agents` to `agent-loop` (default `[]`), so each app supplies its own pre-created agents — a stdio app pre-creates a `main`; the ACP app pre-creates none (it creates agents on demand at `session/new`) — `maxParallelToolCalls` to `agent-loop` as the shared concurrent tool-call cap for every agent it creates; `persona` and `toolOrder` to `dsh-system-prompt`; `tools` to the tool registry for its presentation mode; and `skills.registry`, `skills.local`, and `skills.tool` to the skill registry, local provider, and model-facing consumer. Forwarding is exactly why the owners can live in the shared spine even though the apps disagree on what to configure.
## Why a code bundle, not a shared YAML include

4
packages/core/agent-core/src/index.ts
View File

@@ -47,8 +47,8 @@ export interface Config {
/** The agent-loop `agents` list (see dsh-agent-loop's `Config`). */
agents?: AgentLoopConfig['agents']
/**
* The factory-wide default concurrent tool-call cap applied to every agent
* this bundle creates (see dsh-agent-loop's `Config.maxParallelToolCalls`).
* Concurrent tool-call cap shared by every agent this bundle's loop creates
* (see dsh-agent-loop's `Config.maxParallelToolCalls`).
*/
maxParallelToolCalls?: AgentLoopConfig['maxParallelToolCalls']
/** The deployment persona (see dsh-system-prompt's `Config`). */

5
packages/core/agent-core/tests/agent-core.spec.ts
View File

@@ -117,13 +117,12 @@ describe('dsh-agent-core bundle', () => {
await ctx.fiber.dispose()
})
it('forwards the factory-wide maxParallelToolCalls default to agents without a per-agent cap', async () => {
it('forwards the global maxParallelToolCalls config to agent-loop', async () => {
const ctx = await mount({
agents: [{ id: AgentId('main'), model: 'mock' }],
maxParallelToolCalls: 3,
})
const main = ctx.get('agents')?.get(AgentId('main'))
expect(main?.options.maxParallelToolCalls).toBe(3)
expect(ctx.get('agentLoop')?.config.maxParallelToolCalls).toBe(3)
await ctx.fiber.dispose()
})

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

@@ -29,17 +29,17 @@ The config-driven `ctx.agentLoop.create()` path keeps its agent owned by the loo
```ts
interface Config {
maxParallelToolCalls?: number // shared by every agent; default 10; 1 is serial
agents: Array<{
id: string // required
model?: string
maxParallelToolCalls?: number // positive integer; default 10; 1 is serial
resumeSessionId?: string // load this persisted session instead of creating one
cwd?: string // optional workspace cwd for the fresh session
}>
}
```
Configured agents start automatically. `cwd` applies only to fresh sessions; `resumeSessionId` retains persisted metadata. `maxParallelToolCalls` bounds the rolling pool for parallel-safe calls and defaults to `10`. They use the deployment persona. Programmatic setup can shadow it per agent. This plugin supplies the per-agent `model` and `cwd` prompt variables; harness identity and deployment persona belong to `dsh-system-prompt`.
Configured agents start automatically. `cwd` applies only to fresh sessions; `resumeSessionId` retains persisted metadata. `maxParallelToolCalls` bounds every agent's rolling pool for parallel-safe calls and defaults to `10`. They use the deployment persona, which programmatic setup can shadow per agent. This plugin supplies the per-agent `model` and `cwd` prompt variables; harness identity and deployment persona belong to `dsh-system-prompt`.
### Exported concrete class

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

@@ -54,15 +54,16 @@ export interface PreparedReactLoopAgent {
* @param id - the concrete agent identity.
* @param options - loop options for the agent.
* @param session - the prepared session the agent will own.
* @param maxParallelToolCalls - resolved scheduler cap shared by this factory's agents.
* @returns the agent and closures bound only to that exact instance.
*/
export function prepareReactLoopAgent(
ctx: Context, id: AgentId, options: AgentOptions, session: Session,
ctx: Context, id: AgentId, options: AgentOptions, session: Session, maxParallelToolCalls: number,
): PreparedReactLoopAgent {
if (claimedDriverSessions.has(session)) {
throw new Error(`session "${session.id}" already has a concrete agent driver`)
}
const agent = new ReactLoopAgent(ctx, id, options, session)
const agent = new ReactLoopAgent(ctx, id, options, session, maxParallelToolCalls)
claimedDriverSessions.add(session)
const dispose = () => agent[stopDriver]()
return {
@@ -143,6 +144,8 @@ 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. */
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
@@ -155,7 +158,9 @@ export class ReactLoopAgent implements Agent {
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
@@ -329,6 +334,7 @@ export class ReactLoopAgent implements Agent {
this.driverStarted = true
this.done = runLoop(this.loopCtx, this, {
inbox: this.#inbox,
maxParallelToolCalls: this.maxParallelToolCalls,
setStatus: (status) => { this.setStatus(status) },
setAbort: controller => void (this.currentAbort = controller),
disposed: this.disposed,

8
packages/core/agent-loop/src/constants.ts
View File

@@ -7,9 +7,9 @@
*/
/**
* 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.
* 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.
*/
export const DEFAULT_MAX_PARALLEL_TOOL_CALLS = 10

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

@@ -32,6 +32,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'
@@ -73,12 +74,13 @@ function signalAbortError(id: AgentId, signal: AbortSignal): Error {
return new Error(`agent "${id}" creation aborted`, { cause: signal.reason })
}
/** Validate merge-extended options the loop owns before a session is published. */
function validateAgentOptions(options: AgentOptions): void {
const { maxParallelToolCalls } = options
if (maxParallelToolCalls !== undefined && (!Number.isInteger(maxParallelToolCalls) || maxParallelToolCalls < 1)) {
/** 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
}
/**
@@ -171,13 +173,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)
@@ -326,32 +328,14 @@ 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 }
export { DEFAULT_MAX_PARALLEL_TOOL_CALLS } from './constants.ts'
/** Plugin configuration for declarative startup agents. */
/** Agent-loop plugin configuration. */
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.
* 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}.
*/
maxParallelToolCalls?: number
/** Agents created or resumed at plugin startup. */
@@ -360,11 +344,6 @@ export interface Config {
id: AgentId
/** Optional workspace for a 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
/** Persisted session to resume instead of creating a fresh session. */
resumeSessionId?: SessionId
})[]
@@ -376,26 +355,25 @@ 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),
// 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(),
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>
private readonly ownership: FactoryOwnership
/** Resolved immutable scheduler cap shared by every driver from 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()')
@@ -423,22 +401,6 @@ 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.
@@ -448,14 +410,12 @@ export class AgentLoop extends Service implements AgentFactory {
* @returns the published running agent.
*/
create(id: AgentId, options: AgentOptions = {}, meta: Pick<SessionHeader, 'cwd'> = {}): ReactLoopAgent {
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(resolved, session)
const agent = transaction.prepare(options, session, this.maxParallelToolCalls)
transaction.publish('startup')
return agent
} catch (error: unknown) {
@@ -473,8 +433,7 @@ export class AgentLoop extends Service implements AgentFactory {
* @returns the published handle.
*/
async createAgent(ownerCtx: Context, options: CreateAgentOptions): Promise<AgentHandle> {
const agentOptions = this.withFactoryDefaults(options.agentOptions ?? {})
validateAgentOptions(agentOptions)
const agentOptions = options.agentOptions ?? {}
const transaction = new AgentCreationTransaction(
this.runtime.ctx,
ownerCtx,
@@ -487,7 +446,7 @@ export class AgentLoop extends Service implements AgentFactory {
...options.seed === undefined ? {} : { seed: options.seed },
...options.meta === undefined ? {} : { meta: options.meta },
})
const agent = transaction.prepare(agentOptions, session)
const agent = transaction.prepare(agentOptions, session, this.maxParallelToolCalls)
await transaction.waitFor(options.setup?.(agent.ctx))
transaction.assertActive()
return transaction.publish('startup')
@@ -519,8 +478,7 @@ export class AgentLoop extends Service implements AgentFactory {
persistence: SessionPersistence,
options: ResumeAgentOptions,
): Promise<AgentHandle> {
const agentOptions = this.withFactoryDefaults(options.agentOptions ?? {})
validateAgentOptions(agentOptions)
const agentOptions = options.agentOptions ?? {}
const transaction = new AgentCreationTransaction(
this.runtime.ctx,
ownerCtx,
@@ -540,7 +498,7 @@ export class AgentLoop extends Service implements AgentFactory {
...loaded.meta.seedLength === undefined ? {} : { seedLength: loaded.meta.seedLength },
},
})
const agent = transaction.prepare(agentOptions, session)
const agent = transaction.prepare(agentOptions, session, this.maxParallelToolCalls)
await transaction.waitFor(options.setup?.(agent.ctx))
transaction.assertActive()
return transaction.publish('resume')

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

@@ -17,7 +17,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, resolveMaxParallelToolCalls } from './tool-calls.ts'
import { executeToolCalls } from './tool-calls.ts'
import type { ReactLoopAgent } from './agent.ts'
import type { Inbox } from './inbox.ts'
@@ -73,6 +73,8 @@ function stepFinishReason(finish: FinishReason): TurnEndReason | undefined {
export interface LoopHandle {
/** Native-private agent inbox handed to the driver only at internal startup. */
readonly inbox: Inbox
/** Immutable concurrent tool-call cap resolved by the owning factory. */
readonly maxParallelToolCalls: number
setStatus(status: 'idle' | 'running'): void
setAbort(controller: AbortController | undefined): void
/** Resolves when the agent is disposed — unblocks the idle wait. */
@@ -326,7 +328,8 @@ async function runTurn(
let stepOutcome: { hadToolCalls: boolean; finish: FinishReason } | { error: Error }
try {
stepOutcome = await runStep(
ctx, events, agent, turn, step, assembly, fullSystemPrompt, boundaryMessages, transmission, abort.signal)
ctx, events, agent, turn, step, assembly, fullSystemPrompt, boundaryMessages,
transmission, abort.signal, handle.maxParallelToolCalls)
} catch (error: unknown) {
stepOutcome = { error: toError(error) }
} finally {
@@ -470,6 +473,7 @@ async function runStep(
boundaryMessages: Message[],
transmission: TransmissionLog,
signal: AbortSignal,
maxParallelToolCalls: number,
): Promise<{ hadToolCalls: boolean; finish: FinishReason }> {
const { session, options } = agent
@@ -545,12 +549,7 @@ async function runStep(
let message: Message = assembler.message()
message = await events.waterfall('agent/step-result', turn, step, message, () => Promise.resolve(message))
// Validate the live cap before logging model-visible tool calls so bad mutable
// options cannot leave unanswered calls in the transcript.
const toolCalls = message.content.filter(block => block.type === 'tool-call')
const maxParallel = toolCalls.length > 0
? resolveMaxParallelToolCalls(agent.options.maxParallelToolCalls)
: undefined
// Empty messages exist only to carry usage; omit empty provenance.
if (message.content.length > 0 || assembler.usage) {
@@ -563,8 +562,8 @@ async function runStep(
// The scheduler overlaps only dispatch/body for parallel-safe calls; policy,
// results, and additional context remain in model order.
const pendingContext = maxParallel !== undefined
? await executeToolCalls(ctx, agent, turn, step, toolCalls, signal, maxParallel)
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.

28
packages/core/agent-loop/src/tool-calls.ts
View File

@@ -4,8 +4,8 @@
* 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 runs every group through the same rolling pool
* bounded by the agent's `maxParallelToolCalls` — an exclusive group is a pool
* of one.
* bounded by the agent-loop's `maxParallelToolCalls` config — 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
@@ -25,7 +25,6 @@ import type { HookContext } from '@deepseek-ai/dsh-agent'
import type { Session } from '@deepseek-ai/dsh-session'
import { TOOL_REGISTRY_SCHEDULER, type ToolExecution, type ToolExecutionInput, 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 {
@@ -107,20 +106,6 @@ export async function executeToolCalls(
return pendingContext
}
/**
* Resolve and validate the per-step parallel dispatch cap before the assistant
* tool-call message is logged, so invalid mutable options fail without leaving
* dangling model-visible tool calls in the session transcript.
*
* @param maxParallelToolCalls - the live agent option value.
* @returns the positive integer cap to use for this step.
*/
export function resolveMaxParallelToolCalls(maxParallelToolCalls: number | undefined): number {
const maxParallel = maxParallelToolCalls ?? DEFAULT_MAX_PARALLEL_TOOL_CALLS
assertMaxParallelToolCalls(maxParallel)
return maxParallel
}
/** Parse a model-produced raw arguments string, falling back to the raw string on invalid JSON (empty ⇒ `{}`). */
function parseArguments(raw: string): unknown {
try {
@@ -157,13 +142,6 @@ function groupByMode(ctx: Context, planned: PlannedCall[]): PlannedCall[][] {
return groups
}
/** Validate the live per-agent cap at the point it controls dispatch. */
function assertMaxParallelToolCalls(maxParallel: number): void {
if (!Number.isInteger(maxParallel) || maxParallel < 1) {
throw new Error('maxParallelToolCalls must be a positive integer')
}
}
/**
* 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
@@ -189,8 +167,6 @@ async function runGroup(
): Promise<void> {
/* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
assertMaxParallelToolCalls(maxParallel)
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,

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

@@ -6,7 +6,7 @@ import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry from '@deepseek-ai/dsh-tools'
import AgentRegistry from '@deepseek-ai/dsh-agent'
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'
@@ -53,10 +53,14 @@ describe('ReactLoopAgent', () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const session = ctx.sessions.create(SessionId('exclusive-driver'))
const prepared = prepareReactLoopAgent(ctx, AgentId('first-driver'), { model: 'mock' }, session)
const prepared = prepareReactLoopAgent(
ctx, AgentId('first-driver'), { model: 'mock' }, session, DEFAULT_MAX_PARALLEL_TOOL_CALLS,
)
expect(() => prepared.agent.ctx).toThrow('context is not bound')
expect(() => prepareReactLoopAgent(ctx, AgentId('second-driver'), { model: 'mock' }, session))
expect(() => prepareReactLoopAgent(
ctx, AgentId('second-driver'), { model: 'mock' }, session, DEFAULT_MAX_PARALLEL_TOOL_CALLS,
))
.toThrow('already has a concrete agent driver')
await prepared.dispose()
@@ -254,7 +258,9 @@ describe('ReactLoopAgent', () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const session = ctx.sessions.create(SessionId('test'))
const prepared = prepareReactLoopAgent(ctx, AgentId('bare'), { model: 'mock' }, session)
const prepared = prepareReactLoopAgent(
ctx, AgentId('bare'), { model: 'mock' }, session, DEFAULT_MAX_PARALLEL_TOOL_CALLS,
)
const { agent } = prepared
prepared.markPublished()
@@ -272,7 +278,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'), { model: 'mock' }, session)
const prepared = prepareReactLoopAgent(
ctx, AgentId('pre-start-dispose'), { model: 'mock' }, session, DEFAULT_MAX_PARALLEL_TOOL_CALLS,
)
await prepared.dispose()
expect(prepared.agent.status).toBe('disposed')
@@ -369,7 +377,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'), { model: 'mock' }, session)
const prepared = prepareReactLoopAgent(
ctx, AgentId('bare'), { model: 'mock' }, session, DEFAULT_MAX_PARALLEL_TOOL_CALLS,
)
const { agent } = prepared
prepared.markPublished()
const dispose = prepared.startDriver()

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

@@ -5,7 +5,7 @@ import SessionStore, { Session, SessionEvent, SessionId, TurnEndReason } from '@
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry, { defineTool } from '@deepseek-ai/dsh-tools'
import AgentRegistry, { AgentId, type ContinuationDecision } from '@deepseek-ai/dsh-agent'
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 { MockAdapter, textResponse, toolCallResponse } from './mock-adapter.ts'
@@ -528,7 +528,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'), { model: 'mock' }, seeded)
const prepared = prepareReactLoopAgent(
ctx2, AgentId('forked-agent'), { model: 'mock' }, seeded, DEFAULT_MAX_PARALLEL_TOOL_CALLS,
)
const forked = prepared.agent
prepared.markPublished()
ctx2.effect(() => prepared.startDriver())

91
packages/core/agent-loop/tests/tool-calls.spec.ts
View File

@@ -12,7 +12,7 @@
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 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'
@@ -20,14 +20,17 @@ 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) {
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(AgentLoop, { agents: [] })
await ctx.plugin(AgentLoop, {
agents: [],
...maxParallelToolCalls === undefined ? {} : { maxParallelToolCalls },
})
ctx.llm.registerAdapter(['mock'], adapter)
return ctx
}
@@ -190,38 +193,28 @@ describe('tool-call scheduler: model-order results despite out-of-order settleme
})
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')
await expect(ctx.agents.create({
agentId: AgentId('bad-fractional'),
sessionId: SessionId('bad-fractional-session'),
agentOptions: { model: 'mock', maxParallelToolCalls: 1.5 },
})).rejects.toThrow('maxParallelToolCalls must be a positive integer')
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('fails loud if maxParallelToolCalls is mutated invalid after agent creation', async () => {
const adapter = new MockAdapter([
multiCall([{ id: 'c1', name: 'p', args: { id: '1' } }, { id: 'c2', name: 'p', args: { id: '2' } }]),
textResponse('must not run after unanswered tool calls'),
])
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.options as { maxParallelToolCalls: number }).maxParallelToolCalls = 0
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')
})
agent.send([{ type: 'text', text: 'go' }])
await waitForIdle(ctx, agent)
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(gated.started).toEqual([])
expect(adapter.requests).toHaveLength(1)
expect(events(agent).some(e => e.type === 'assistant/message')).toBe(false)
expect(events(agent).filter(e => e.type === 'tool/call' || e.type === 'tool/result')).toEqual([])
const turnEnd = events(agent).findLast(e => e.type === 'turn/end')
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason.kind).toBe('error')
expect(() => new AgentLoop(ctx, { agents: [] })).not.toThrow()
await ctx.fiber.dispose()
})
it('starts at most the cap, replenishing as calls settle', async () => {
@@ -229,10 +222,10 @@ describe('tool-call scheduler: rolling pool honors maxParallelToolCalls', () =>
multiCall([1, 2, 3, 4].map(n => ({ id: `c${n}`, name: 'p', args: { id: String(n) } }))),
textResponse('done'),
])
const ctx = await harness(adapter)
const ctx = await harness(adapter, 2)
const gated = gatedParallelTool('p')
ctx.tools.register(gated.tool)
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock', maxParallelToolCalls: 2 })
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
agent.send([{ type: 'text', text: 'go' }])
// Only 2 start initially (the cap).
@@ -261,10 +254,10 @@ describe('tool-call scheduler: rolling pool honors maxParallelToolCalls', () =>
multiCall([{ id: 'c1', name: 'p', args: { id: '1' } }, { id: 'c2', name: 'p', args: { id: '2' } }]),
textResponse('done'),
])
const ctx = await harness(adapter)
const ctx = await harness(adapter, 1)
const gated = gatedParallelTool('p')
ctx.tools.register(gated.tool)
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock', maxParallelToolCalls: 1 })
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
agent.send([{ type: 'text', text: 'go' }])
await until(() => gated.started.length === 1)
await new Promise(r => setTimeout(r, 5))
@@ -275,7 +268,7 @@ describe('tool-call scheduler: rolling pool honors maxParallelToolCalls', () =>
await waitForIdle(ctx, agent)
})
it('applies the factory-wide Config default to agents that set no per-agent cap', async () => {
it('applies the global Config cap to every agent created by the factory', async () => {
const adapter = new MockAdapter([
multiCall([{ id: 'c1', name: 'p', args: { id: '1' } }, { id: 'c2', name: 'p', args: { id: '2' } }]),
textResponse('done'),
@@ -286,14 +279,12 @@ describe('tool-call scheduler: rolling pool honors maxParallelToolCalls', () =>
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.
// The global cap of 1 must serialize every agent from this factory.
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))
@@ -304,18 +295,6 @@ describe('tool-call scheduler: rolling pool honors maxParallelToolCalls', () =>
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', () => {
@@ -349,7 +328,7 @@ describe('tool-call scheduler: ordered middleware and additionalContext', () =>
multiCall([{ id: 'c1', name: 'p', args: { id: '1' } }, { id: 'c2', name: 'p', args: { id: '2' } }]),
textResponse('done'),
])
const ctx = await harness(adapter)
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> =>
@@ -467,14 +446,14 @@ describe('tool-call scheduler: abort handling', () => {
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 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(),
additionalContext: { content: [{ type: 'text', text: `ctx-${exec.callId}` }], source: { kind: 'plugin', plugin: 'p' } },
}))
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock', maxParallelToolCalls: 2 })
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
agent.send([{ type: 'text', text: 'go' }])
await until(() => gated.started.length === 2)
@@ -500,7 +479,7 @@ describe('tool-call scheduler: abort handling', () => {
]),
textResponse('should never be requested'),
])
const ctx = await harness(adapter)
const ctx = await harness(adapter, 2)
const gated = gatedParallelTool('p')
const exclusive: string[] = []
ctx.tools.register(gated.tool)
@@ -510,7 +489,7 @@ describe('tool-call scheduler: abort handling', () => {
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 })
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
agent.send([{ type: 'text', text: 'go' }])
await until(() => gated.started.length === 2)

1
packages/ui/acp/README.md
View File

@@ -15,7 +15,6 @@ The plugin injects `agents`, `sessions`, `sessionPersistence`, `tools`, and `use
| Key | Default | Meaning |
|---|---|---|
| `model` | — | Model name for created agents (must have a registered adapter). |
| `maxParallelToolCalls` | (agent-loop default) | Positive integer cap on tool calls each created agent runs concurrently within one assistant step; `1` is fully serial. |
(No persona key: `dsh-system-prompt`'s own `persona` config supplies the global default section, so ACP-created agents render it without the bridge carrying prompt text. An agent-scoped same-name section may still shadow that default.)

12
packages/ui/acp/src/index.ts
View File

@@ -203,17 +203,12 @@ function stringArrayContent(
export interface AcpConfig {
/** Model name for created agents (must have a registered adapter). */
model?: string
/** Positive-integer concurrent tool-call cap for each created agent; `1` is serial. */
maxParallelToolCalls?: number
/** Runtime-only transport override for tests; production uses stdio. */
stream?: Stream
}
export const Config: Schema<AcpConfig> = Schema.object({
model: Schema.string(),
// A positive integer; a bad value (0, negative, fractional) fails config
// validation here rather than being silently dropped from cordis.yml.
maxParallelToolCalls: Schema.number().step(1).min(1),
})
/** Per-session bridge state keyed by ACP session id. */
@@ -856,13 +851,12 @@ export function apply(ctx: Context, config: AcpConfig): void {
* Build per-agent options from the plugin config, omitting absent fields
* (exactOptionalPropertyTypes: never assign `undefined` to an optional key).
* Exported for unit coverage of both the present and absent branches.
* @param config - the plugin config carrying the optional model name and parallel cap.
* @returns the per-agent options, with each field present only when configured.
* @param config - the plugin config carrying the optional model name.
* @returns the per-agent options, with `model` present only when configured.
*/
export function agentOptions(config: AcpConfig): { model?: string; maxParallelToolCalls?: number } {
export function agentOptions(config: AcpConfig): { model?: string } {
return {
...config.model !== undefined ? { model: config.model } : {},
...config.maxParallelToolCalls !== undefined ? { maxParallelToolCalls: config.maxParallelToolCalls } : {},
}
}

2
packages/ui/acp/tests/stream-update.spec.ts
View File

@@ -794,7 +794,5 @@ describe('agentOptions', () => {
it('includes only the fields present in config', () => {
expect(agentOptions({})).toEqual({})
expect(agentOptions({ model: 'm' })).toEqual({ model: 'm' })
expect(agentOptions({ maxParallelToolCalls: 3 })).toEqual({ maxParallelToolCalls: 3 })
expect(agentOptions({ model: 'm', maxParallelToolCalls: 1 })).toEqual({ model: 'm', maxParallelToolCalls: 1 })
})
})

2
packages/ui/stdio-agent/README.md
View File

@@ -26,7 +26,7 @@ The leaf `cordis.yml` supplies only the **swappable backends** — an LLM adapte
| Key | Default | Routed to |
|---|---|---|
| `model` | (required) | the pre-created `main` agent's model |
| `maxParallelToolCalls` | agent-loop default | positive-integer concurrent tool-call cap for the `main` agent; `1` is serial |
| `maxParallelToolCalls` | agent-loop default | positive-integer concurrent tool-call cap shared by the bundled loop's agents; `1` is serial |
| `persona` | — | the deployment persona template (may reference `{{model}}`/`{{cwd}}`), routed to `dsh-system-prompt` |
| `toolOrder` | — | explicit model-facing tool order (a name list with one `'<unlisted-tools>'` rest entry; absent — lexicographic; an unregistered name fails each turn at prompt assembly), routed to `dsh-system-prompt` |
| `tools` | `{ mode: 'native' }` | tool-registry presentation config (`native` / `code` / `both`), routed through `dsh-agent-core` |

7
packages/ui/stdio-agent/src/index.ts
View File

@@ -37,9 +37,8 @@ export interface Config {
/** Model name for the `main` agent (must have a registered adapter). */
model: string
/**
* Maximum tool calls the `main` agent runs concurrently within one assistant
* step (a positive integer; the agent loop defaults it when omitted). `1`
* preserves fully serial execution.
* Concurrent parallel-safe tool-call cap for the bundled agent loop. A
* positive integer; the loop defaults it when omitted and `1` is serial.
*/
maxParallelToolCalls?: number
/** Deployment persona (the system-prompt plugin's `persona` config). */
@@ -94,11 +93,11 @@ export function apply(ctx: Context, config: Config): void {
...config.persona !== undefined ? { persona: config.persona } : {},
...config.toolOrder !== undefined ? { toolOrder: config.toolOrder } : {},
...config.tools !== undefined ? { tools: config.tools } : {},
...config.maxParallelToolCalls !== undefined ? { maxParallelToolCalls: config.maxParallelToolCalls } : {},
agents: [{
id: AgentId('main'),
model: config.model,
cwd: process.cwd(),
...config.maxParallelToolCalls !== undefined ? { maxParallelToolCalls: config.maxParallelToolCalls } : {},
...config.resumeSessionId !== undefined ? { resumeSessionId: SessionId(config.resumeSessionId) } : {},
}],
...config.skills !== undefined ? { skills: config.skills } : {},

4
packages/ui/stdio-agent/tests/stdio-agent.spec.ts
View File

@@ -126,14 +126,14 @@ describe('dsh-stdio-agent app', () => {
await ctx.fiber.dispose()
})
it('forwards maxParallelToolCalls onto the pre-created agent when set', async () => {
it('forwards maxParallelToolCalls to the bundled agent loop', async () => {
const ctx = await mount({
model: 'mock',
maxParallelToolCalls: 3,
persistenceRoot: '/tmp/dsh-stdio-agent-spec-parallel',
skills: await isolatedSkillsConfig(),
})
expect(ctx.get('agents')?.get(AgentId('main'))?.options.maxParallelToolCalls).toBe(3)
expect(ctx.get('agentLoop')?.config.maxParallelToolCalls).toBe(3)
await ctx.fiber.dispose()
})