diff --git a/docs/config-catalog.md b/docs/config-catalog.md index dbee6b350f..abe03d6461 100644 --- a/docs/config-catalog.md +++ b/docs/config-catalog.md @@ -18,8 +18,6 @@ Requires: `agents` · `sessions` · `sessionPersistence` · `tools` · `userInte 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 } @@ -77,8 +75,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`). */ @@ -111,16 +109,12 @@ Source: [`packages/core/agent-core/src/index.ts:46`](../packages/core/agent-core Requires: `agents` · `sessions` · `llm` · `tools` · `systemPrompt` ```ts config-catalog -/** 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. */ @@ -129,11 +123,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 })[] @@ -142,7 +131,7 @@ export interface Config { Depends on: [`AgentId`](../packages/core/agent/src/index.ts) · [`AgentOptions`](../packages/core/agent/src/index.ts) · [`SessionId`](../packages/core/session/src/index.ts) -Source: [`packages/core/agent-loop/src/index.ts:346`](../packages/core/agent-loop/src/index.ts) +Source: [`packages/core/agent-loop/src/index.ts:334`](../packages/core/agent-loop/src/index.ts) ## `@deepseek-ai/dsh-bash-local` @@ -647,9 +636,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). */ diff --git a/docs/cordis-catalog/services.md b/docs/cordis-catalog/services.md index 830184e804..12a41b2aed 100644 --- a/docs/cordis-catalog/services.md +++ b/docs/cordis-catalog/services.md @@ -19,7 +19,7 @@ async createAgent(ownerCtx: Context, options: CreateAgentOptions): Promise ``` -Source: [`packages/core/agent-loop/src/index.ts:374`](../../packages/core/agent-loop/src/index.ts) +Source: [`packages/core/agent-loop/src/index.ts:353`](../../packages/core/agent-loop/src/index.ts) ## `ctx.agents` — `AgentRegistry` diff --git a/docs/core-data-structures/core.md b/docs/core-data-structures/core.md index 42c11a84f0..9a9a1706fc 100644 --- a/docs/core-data-structures/core.md +++ b/docs/core-data-structures/core.md @@ -347,7 +347,7 @@ interface Agent { } ``` -`AgentStatus` is `'idle' | 'running' | 'disposed'`, and `AgentId` is branded. `AgentOptions` is merge-extensible and currently includes `model?` and `maxParallelToolCalls?` (the loop's per-agent concurrent tool-call cap; the owning field, defaulted by `AgentLoop.Config` and, absent that, by `DEFAULT_MAX_PARALLEL_TOOL_CALLS`). Persona belongs to `dsh-system-prompt`: an agent-scoped `deployment:persona` may shadow the global default. +`AgentStatus` is `'idle' | 'running' | 'disposed'`, and `AgentId` is branded. `AgentOptions` is merge-extensible and currently includes `model?`. Persona belongs to `dsh-system-prompt`: an agent-scoped `deployment:persona` may shadow the global default. The [event taxonomy](../architecture.md#event) owns the `agent/*` lifecycle, checkpoint, and waterfall contracts. Turn and step boundaries are durable session events rather than agent emits. diff --git a/docs/rfc/implemented/feature/2026-07-10-parallel-tool-call-execution.md b/docs/rfc/implemented/feature/2026-07-10-parallel-tool-call-execution.md index 23e0f0d22e..74a8ef6c61 100644 --- a/docs/rfc/implemented/feature/2026-07-10-parallel-tool-call-execution.md +++ b/docs/rfc/implemented/feature/2026-07-10-parallel-tool-call-execution.md @@ -4,109 +4,100 @@ Status: implemented ## Problem -The loop accepts an assistant message containing multiple `tool-call` blocks. Serial execution makes independent reads, web requests, and subagent delegations pay the sum of their wall-clock latency even though the model and adapters already represent sibling tool calls in one response. +An assistant message may contain several sibling `tool-call` blocks. Running them serially adds the latency of independent reads, web requests, and subagent runs even though the model has already requested them together. -Concurrency cannot live in the model-facing JSON schema. `ctx.tools.schemas()` exposes only `name`, `description`, and `parameters`; scheduling is a host contract. The loop needs an internal per-call safety decision and must use it without hardcoding tool names. +Concurrency is a host scheduling concern, not model-facing tool metadata. The loop needs to decide which calls may overlap without hardcoding tool names or exposing scheduler policy in the JSON schema. -The hard constraint is replay. The session log remains the source of truth: the assistant message contains the model's calls in order, each started call has a `tool/call` audit event before its body runs, each model-facing result is a `tool/result`, and derived history sees results in the original call order. Live ACP and stdio surfaces may show several pending calls before the first result; that progress interleaving is not part of the model-history guarantee. +The session log remains authoritative: every started call has an audit event, every started call receives a result, and model history observes results in the original call order regardless of completion order. ## Decision -`ToolDefinition` carries an optional host-only classifier: +Each tool may provide an optional `isConcurrencySafe(args)` classifier. It is synchronous and pure: it examines only the current call's parsed arguments and performs no I/O or mutation. Only an explicit `true` opts in; a missing classifier, invalid arguments, a thrown classifier, or any other return value makes the call exclusive. The canonical type contract lives in the [tool data structures](../../../core-data-structures/tools.md). + +The classifier is deliberately unary. Returning `true` is the tool's promise that this call may overlap with any sibling call that also returns `true`; the scheduler does not compare calls or prove that their resource accesses are compatible. + +Arguments still support input-sensitive classification. A tool may classify a read-only operation as parallel and a mutating operation as exclusive. The interface cannot express relational rules such as "these writes are safe only when their paths differ," so a call whose safety depends on a sibling remains exclusive. + +`defineTool()` validates arguments before invoking a typed classifier. Invalid arguments classify as exclusive and produce the ordinary argument error only if the call executes. `ctx.tools.executionMode(exec)` resolves the live tool definition and returns the tagged `parallel` or `exclusive` mode; unknown tools fail closed to exclusive. + +A tagged mode, rather than a public boolean scheduler API, leaves room for a future resource-aware mode without changing the classifier contract. + +## Scheduling and ordering + +The loop waits for the complete assistant message, parses every call once, creates a distinct `ToolExecution` for each call, and scans them in model order. Consecutive parallel calls form one group; every exclusive call forms a singleton group and an ordering barrier. Groups execute sequentially. + +For example: ```text -export interface ToolDefinition extends ToolSchema { - execute(args: unknown, exec: ToolExecution): Promise - isConcurrencySafe?(args: unknown): boolean -} +[parallel read(A), parallel read(B), exclusive write(A), parallel read(C)] + +→ [read(A), read(B)] +→ [write(A)] +→ [read(C)] ``` -`isConcurrencySafe` is synchronous, pure classification metadata. It may inspect parsed call arguments; `defineTool()` schema-validates those arguments before the typed callback runs, while hand-rolled definitions receive the raw parsed value. The callback performs no I/O and receives no live `Agent` or mutable `ToolExecution`. `defineTool()` validates arguments softly for `isConcurrencySafe`, matching the display-only `presentCall`/`presentResult` pattern: invalid args return `false`, and the ordinary `ToolArgsError` is produced only if the tool executes. +`read(A)` and `read(B)` may overlap. `write(A)` starts after both finish, and `read(C)` starts after the write finishes. -The registry exposes the scheduling decision as a plain method: +Every group uses a rolling pool bounded by `maxParallelToolCalls`: the loop starts calls in model order up to the cap and starts another whenever one settles. An exclusive group is a pool of one. A cap of `1` preserves serial execution. -```text -export type ToolExecutionMode = - | { kind: 'parallel' } - | { kind: 'exclusive' } -``` +Only dispatch and the tool body overlap. `tools/pre-execute` and `tools/post-execute` run in model order because middleware may maintain ordering-sensitive state. `tools/execute` wrappers run around concurrent dispatches and therefore must be reentrant across distinct executions. -```text -class ToolRegistry { - executionMode(exec: ToolExecutionInput): ToolExecutionMode -} -``` +Each started call appends `tool/call` immediately before its pre-execute gate. Completed dispatches occupy model-order slots, and a commit cursor appends `tool/result` and collects `additionalContext` only when the next slot is ready. Live surfaces may show several pending calls, but results and post-tool context remain model-ordered. -`ctx.tools.executionMode(exec)` looks up the registered tool and calls `tool.isConcurrencySafe?.(exec.arguments)`. Unknown tools, missing declarations, malformed typed args, and thrown safety checks all resolve to `{ kind: 'exclusive' }`. The method is not a Cordis waterfall; it is the future insertion point if hook, MCP, or provider policy needs to downgrade a tool's baseline decision. The object-tagged union leaves room for future resource grouping, for example `{ kind: 'exclusive', group: 'session:...' }`. +An abort before a group starts records no calls from that group. An abort during a group stops replenishment, waits for already-started calls, commits their results in order, drops their buffered additional context, and then ends the step through the existing abort path. Calls that never start have no audit event. -A parallel-safe declaration is a contract. The tool body must not mutate the parent agent's session or other parent-owned async state during `execute`; parent-session writes such as `exec.agent.session.append(...)`, `agent.inject(...)`, or other tool-owned parent events belong to exclusive tools unless the mutation moves behind the loop's ordered result path. The only parent-step outputs a parallel-safe call may produce are its returned content, `meta`, structured error, and `additionalContext` carried through the ordered post-execute path. The narrow exception is a synchronous, side-effect-only recorder whose updates are commutative or fail closed for concurrent calls by the same session. `fs/observed` is the worked example: `read` emits it synchronously after a successful read, `dsh-fs-policy` records `WeakMap` state synchronously, and write/edit remain exclusive barriers that re-check versions before mutating; a stale observation can only make the provider CAS reject with `FS_STALE_VERSION`. +Code Mode remains outside this scheduler because the model emits one native `run_code` call. `run_code` and its internal dispatch queue remain serial; native sibling calls in `mode: 'both'` use the normal scheduler. -## Scheduling +## Safety contract -The loop waits for the model stream to finish and logs one authoritative `assistant/message` before scheduling tools. Streaming tool execution is out of scope. +A tool that returns `true` promises that its body is safe to run at the same time as other parallel calls. It must not directly mutate the parent session or other parent-owned state; it returns its outputs to the loop, which commits them in model order. -For each assistant step, `packages/core/agent-loop/src/tool-calls.ts` parses each call's raw JSON arguments exactly once, creates one distinct `ToolExecution` object per call, asks `ctx.tools.executionMode(exec)`, and partitions calls into ordered groups. A group is either one exclusive call or a run of consecutive parallel calls; grouping classifies each call exactly once. `loop.ts` calls the helper so the turn/step lifecycle remains readable. +Any shared state touched during execution must be concurrency-safe. This includes tool wrappers and providers: they may serialize internally or enforce their own capacity, but they must support concurrent dispatch without corrupting state. -Parallelism is per agent. `AgentOptions.maxParallelToolCalls` is a positive integer, defaults to `DEFAULT_MAX_PARALLEL_TOOL_CALLS` (`10`), and reaches an agent three ways in precedence order: the per-agent option, the factory-wide `AgentLoop.Config.maxParallelToolCalls` applied to every agent the loop creates (declarative startup agents and factory callers such as the ACP, stdio, and SDK front doors), then the built-in default. The factory default is the single `cordis.yml` knob for agents whose front door exposes no cap field of its own. Setting the value to `1` preserves serial execution for that agent. The TypeScript `AgentOptions` vocabulary and both `AgentLoop.Config` fields validate the cap, so invalid `cordis.yml` values fail during config validation. +## Configuration and declarations -Every group runs through the same rolling pool: start calls in model order up to `maxParallelToolCalls`, and whenever one call settles, start the next unstarted call until the group is exhausted. An exclusive group is a pool of one — a barrier — so the loop needs no separate serial path. A group larger than the cap is not truncated; the cap limits simultaneous in-flight calls only. +`maxParallelToolCalls` is a positive AgentLoop deployment cap shared by every agent the factory creates. It defaults to `10`; `1` preserves serial execution. Exact fields and defaults live in the generated [configuration catalog](../../../config-catalog.md). -Only the dispatch/body stage runs concurrently. Generic middleware that can shape ordering-sensitive state remains ordered: `tools/pre-execute` and `tools/post-execute` run in model call order. `@deepseek-ai/dsh-tools` exposes the symbol-keyed internal `TOOL_REGISTRY_SCHEDULER` view so `dsh-agent-loop` can split prepare, dispatch, and finalize without adding named staged service methods to `ctx.tools`; ordinary callers still use the one-call `execute(exec)` API. `tools/execute` around-dispatch listeners run with the dispatch they wrap, so wrappers must be reentrant across distinct `ToolExecution` objects. The shipped timeout policy is per-call: every call owns its mutable `exec` and deadline. +The shipped declarations are conservative. Web search, web fetch, filesystem read, and subagent calls opt in. Filesystem writes and edits, bash tools, workflow, user interaction, todo mutation, Code Mode, and Cordis mutation tools remain exclusive. Bash stays exclusive until its owning package supplies a proven input-sensitive classifier. -Each started call appends its own `tool/call` immediately before its pre-execute gate and body can run. `tool/call` events remain in model order relative to started calls, but their log positions may interleave with sibling results: a later call's `tool/call` can appear before or after an earlier call's `tool/result` as the rolling pool replenishes. That is safe because `tool/call` is log-only; derived model history reads the assistant's `tool-call` blocks and the ordered `tool/result` events, pairing by `callId`. Settled dispatches are stored in model-order slots, and a commit cursor appends `tool/result` only while the next slot is ready. `additionalContext` is collected from those same slots and injected in model call order after normal completion of every started tool result in the step. +Filesystem read relies on a narrow recorder exception: its synchronous observation updates may settle out of order, but write and edit re-check the observed version before mutation, so stale state only produces `FS_STALE_VERSION`. -If the parent signal is already aborted before a group starts, the group is not started and no `tool/call` audit records are appended for it. If the signal aborts while a parallel group is running, the pool stops replenishing, waits for only the already-started calls to settle, records their results in order, drops buffered `additionalContext`, and then raises the abort error so the existing `runTurn` catch path owns `turn/end` reason selection. This keeps every started call paired while avoiding audit records for calls that never began. +The subagent declaration requires providers to accept concurrent `start()` calls for independent runs. A provider may queue, enforce its own capacity, or return a typed failure instead of requiring the parent loop to serialize every subagent call. -Code Mode remains outside native scheduling. In `mode: 'code'`, the wire exposes only `run_code`, so the model emits one native tool call and the loop-level scheduler has nothing to parallelize. `run_code` stays exclusive, and its in-program dispatch queue remains serialized. In `mode: 'both'`, native sibling tool calls can form parallel groups normally, while calls made inside one `run_code` execution still follow Code Mode's own queue. +## Verification -## Tool declarations +Unit coverage pins fail-closed classification, typed argument validation, grouping, barriers, the rolling cap, distinct execution objects, middleware order, ordered results and context, and abort draining. First-party tests pin each parallel declaration. -The shipped declarations are conservative: - -- `web_search`, `web_fetch`, filesystem `read`, and `subagent` return `true`. -- Filesystem `write`, filesystem `edit`, `todo_write`, `bash`, `bash_output`, `bash_kill`, `workflow`, `ask_user_question`, and Cordis mutation tools stay exclusive by omitting `isConcurrencySafe`. -- Bash stays exclusive until a bash-owned read-only classifier exists; the loop never infers shell safety. - -Subagent providers do not get an extra opt-in field. `SubagentProvider.start()` is part of the provider contract and must be safe to call concurrently for independent runs. A provider backed by a limited resource may queue internally, apply its own capacity limit, or return a typed failure for the affected run, but it must not require the parent agent loop to serialize every `subagent` tool call. Built-in spawn, fork, and ACP runs own a child session or process; fork seeds only the parent's completed-turn prefix, so concurrent forks inside the parent's open step all see the same stable prefix. - -Exclusive tools naturally form ordering barriers. A step such as `[read A, write A, read A]` becomes three ordered groups because `write` is exclusive, so the scheduler does not introduce a read/write race inside one assistant step. - -The subagent tool remains synchronous. Multiple subagent tool calls in one assistant message can run concurrently, but each tool result is still the child final answer. Background spawning plus later collection would be a separate tool vocabulary. - -## Testing - -Unit tests cover the classifier (`ToolDefinition.isConcurrencySafe`, `defineTool()` soft validation, `ToolRegistry.executionMode`, and schema projection), the loop scheduler (grouping, exclusive barriers, rolling-pool replenishment, `maxParallelToolCalls: 1`, distinct `ToolExecution` objects, ordered pre/post middleware, ordered `tool/result`, concrete `tool/call`/`tool/result` interleaving, ordered `additionalContext`, and abort/drop-context cases), and first-party safe declarations for filesystem read, web tools, and subagent. - -Snapshot coverage pins the transcript-facing ACP behavior for a multi-call step: several pending tool-call updates may precede model-ordered result updates. Code Mode tests and docs pin that `run_code` remains exclusive and that in-program dispatch stays serialized. No real-API e2e is required for this decision because scheduling is deterministic loop behavior with mocked tools and replayable snapshots, not provider-specific behavior. +Snapshot coverage pins the visible multi-call transcript: pending calls may overlap while completed results remain model-ordered. Code Mode coverage pins its serial boundary. No provider-backed e2e is required because scheduling is deterministic loop behavior. ## Alternatives considered -**Keep serial execution.** This keeps the loop simple and avoids new abort ordering cases, but it leaves obvious latency on the table for independent reads, web calls, and subagent delegations. The model and adapters already represent multiple tool calls in one assistant message, so serial execution is a host limitation rather than a protocol limitation. +**Keep serial execution.** This avoids new ordering and abort cases but retains unnecessary latency for independent sibling calls. -**Codex-style tool-level `supportsParallelToolCalls`.** A tool-level boolean is smaller, but it cannot express that the same tool is safe for some inputs and unsafe for others. Bash is the key example: a read-only command classifier can make `pwd` or `ls` parallel-safe without making `rm` or a long-lived background-task operation parallel-safe. +**Use one tool-level boolean.** A fixed `supportsParallelToolCalls` flag is smaller but cannot distinguish a tool's read-only and mutating operations. The argument-sensitive classifier preserves that distinction. -**Parallelize the complete `ctx.tools.execute()` pipeline.** This preserves the existing one-call API in the loop, but it also runs `tools/pre-execute` and `tools/post-execute` concurrently. The shipped repeat-tool guard and hook bridges can carry ordering-sensitive state, so the shipped design keeps pre/post ordered and overlaps only dispatch/body work. +**Use stateful classification.** Giving the classifier a live agent, registry, or I/O access makes the decision depend on when it runs and creates a gap between classification and dispatch. Mutable authorization and stale-state checks remain execution-time responsibilities. -**Expose a public staged API such as `prepare` / `dispatch` / `finalize`.** That names too much implementation surface before another consumer exists. The loop needs staged behavior, but `ToolRegistry` factors it through a symbol-keyed internal view while keeping `execute(exec)` as the public one-call API for ordinary callers. +**Use sibling-aware or resource-aware classification.** The scheduler could compare calls pairwise or let each call declare resource read/write claims. This can parallelize non-conflicting writes, but it requires shared resource identity and conflict semantics across unrelated tools. The unary contract instead gives up that concurrency and fails closed when safety is relational. -**Add a `tools/execution-mode` waterfall.** A Cordis seam would let hook bridges, provider policies, or MCP server metadata downgrade a tool's declaration. It is not needed for the conservative declaration set: raw and undeclared tools default exclusive, pre/post middleware stays ordered, and a non-reentrant around-dispatch wrapper can serialize internally. The `executionMode(exec)` method remains the insertion point if a real deployment needs policy-driven downgrades. +**Parallelize the complete tool pipeline.** This keeps the loop on the public one-call API but runs pre- and post-execute middleware concurrently. Existing guards and hook bridges may carry ordered state, so only dispatch overlaps. -**Start tools while the model is still streaming.** Claude Code has a streaming executor path, but this repo's log reconstruction and surface-pairing contracts make that a larger design. This decision waits for the assistant message to be assembled, so the log records one authoritative assistant message before scheduling tools. +**Expose staged methods or a scheduling waterfall.** Public `prepare` / `dispatch` / `finalize` methods or a `tools/execution-mode` event add extension surface before another consumer needs it. The loop uses an internal scheduler view, while `executionMode(exec)` remains the insertion point for a future policy seam. -**Use fixed windows inside one parallel group.** Fixed windows would start `maxParallelToolCalls` calls, wait for all of them to settle, then start the next window. The rolling pool wins because slot-based result storage and a model-order commit cursor preserve the transcript contract without sacrificing avoidable latency. +**Start calls while the model streams.** This may reduce latency further but changes assistant-message authority, replay, and call/result pairing. The scheduler starts only after the assistant message is complete. -**Expose concurrency in the model-facing schema.** The model does not need a scheduler flag to request multiple calls; it already can emit multiple `tool-call` blocks. Sending host-only concurrency metadata would bloat requests and mix execution policy into the schema whose job is only argument shape and tool-choice guidance. +**Use fixed-size windows.** Waiting for every call in one window before starting the next leaves capacity idle behind a slow call. The rolling pool preserves the cap without that delay. + +**Expose concurrency metadata to the model.** The model can already emit sibling calls. Host scheduling metadata would enlarge requests without improving tool choice. ## Consequences -Parallel execution can expose latent shared-state bugs in tools that declare themselves safe too broadly. The default is exclusive, the shipped declarations are conservative, and input-sensitive tools such as bash stay exclusive until their owning package proves a narrower classifier. +The design is fail-closed and simple for tool authors, but it cannot exploit concurrency whose safety depends on comparing siblings. A tool that opts in too broadly can expose latent shared-state races. -Tool registration changes are a scheduling boundary. A call classified against one tool definition can become unsafe if an earlier exclusive tool replaces that definition before dispatch, so registry-mutating tools stay exclusive and scheduler changes that cross such barriers must either reclassify against the live registry view or bind dispatch to the classified definition. +Parallel calls may begin in cases where serial execution would have aborted before reaching them. The scheduler therefore records only started calls, drains them on abort, and never starts replacements after cancellation. -An around-dispatch plugin can also violate the contract even when the tool itself is safe. The scheduler limits that risk to `tools/execute`; shipped wrappers are per-call, and third-party wrappers with shared mutable state must serialize internally. +Ordered commits may hold a fast result behind a slow earlier sibling. This preserves replay and model-history order while live surfaces still show pending progress. -Parallel groups change abort timing: a sibling call may have started in a case where the serial loop would not have reached it yet. The pool makes this explicit by logging only started calls, stopping replenishment on abort, draining those calls to results, and preventing later calls from starting. +Concurrent subagents and external calls can compete for quota or process capacity. Providers own their capacity controls; the loop cap only limits calls from one agent step. -Concurrent subagents can compete for model quota, filesystem state, or external process resources. The provider contract requires concurrent `start()` safety, not unlimited capacity, and tool guidance still tells the model to parallelize only independent tasks with non-overlapping write scopes. - -The result-order rule can delay a fast result behind a slow sibling in the same group. That preserves the model transcript and replay contract. ACP and stdio still expose immediate pending-call progress, but completion updates stay model-ordered. +Tool registration is a scheduling boundary. The scheduler currently plans all groups before dispatch, so an earlier registry mutation can make a later classification stale. Binding dispatch to the classified definition or reclassifying after exclusive barriers remains a named correctness gap. diff --git a/packages/core/agent-core/README.md b/packages/core/agent-core/README.md index 594c4a8c82..ea9d94b011 100644 --- a/packages/core/agent-core/README.md +++ b/packages/core/agent-core/README.md @@ -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 diff --git a/packages/core/agent-core/src/index.ts b/packages/core/agent-core/src/index.ts index fcb4e87f1e..7e4da95921 100644 --- a/packages/core/agent-core/src/index.ts +++ b/packages/core/agent-core/src/index.ts @@ -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`). */ diff --git a/packages/core/agent-core/tests/agent-core.spec.ts b/packages/core/agent-core/tests/agent-core.spec.ts index fe240da8ed..4657bb2a40 100644 --- a/packages/core/agent-core/tests/agent-core.spec.ts +++ b/packages/core/agent-core/tests/agent-core.spec.ts @@ -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() }) diff --git a/packages/core/agent-loop/README.md b/packages/core/agent-loop/README.md index e35089b409..1460203cd7 100644 --- a/packages/core/agent-loop/README.md +++ b/packages/core/agent-loop/README.md @@ -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 diff --git a/packages/core/agent-loop/src/agent.ts b/packages/core/agent-loop/src/agent.ts index a2288c65b7..72834db998 100644 --- a/packages/core/agent-loop/src/agent.ts +++ b/packages/core/agent-loop/src/agent.ts @@ -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() 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, diff --git a/packages/core/agent-loop/src/constants.ts b/packages/core/agent-loop/src/constants.ts index de18e0411e..72ba7051c2 100644 --- a/packages/core/agent-loop/src/constants.ts +++ b/packages/core/agent-loop/src/constants.ts @@ -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 diff --git a/packages/core/agent-loop/src/index.ts b/packages/core/agent-loop/src/index.ts index 86df80cd8e..3c9c807e15 100644 --- a/packages/core/agent-loop/src/index.ts +++ b/packages/core/agent-loop/src/index.ts @@ -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() 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 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 = {}): 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 { - 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 { - 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') diff --git a/packages/core/agent-loop/src/loop.ts b/packages/core/agent-loop/src/loop.ts index 7efbceef5e..a04f653b65 100644 --- a/packages/core/agent-loop/src/loop.ts +++ b/packages/core/agent-loop/src/loop.ts @@ -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. diff --git a/packages/core/agent-loop/src/tool-calls.ts b/packages/core/agent-loop/src/tool-calls.ts index d117823a43..52328a080a 100644 --- a/packages/core/agent-loop/src/tool-calls.ts +++ b/packages/core/agent-loop/src/tool-calls.ts @@ -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 { /* 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, diff --git a/packages/core/agent-loop/tests/agent.spec.ts b/packages/core/agent-loop/tests/agent.spec.ts index 581b6fa207..2f3f23bcd2 100644 --- a/packages/core/agent-loop/tests/agent.spec.ts +++ b/packages/core/agent-loop/tests/agent.spec.ts @@ -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() diff --git a/packages/core/agent-loop/tests/contract-regressions.spec.ts b/packages/core/agent-loop/tests/contract-regressions.spec.ts index 8130d4b893..00c5c45c09 100644 --- a/packages/core/agent-loop/tests/contract-regressions.spec.ts +++ b/packages/core/agent-loop/tests/contract-regressions.spec.ts @@ -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()) diff --git a/packages/core/agent-loop/tests/tool-calls.spec.ts b/packages/core/agent-loop/tests/tool-calls.spec.ts index a270ad6da1..8ca6170b03 100644 --- a/packages/core/agent-loop/tests/tool-calls.spec.ts +++ b/packages/core/agent-loop/tests/tool-calls.spec.ts @@ -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 => @@ -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 => ({ ...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) diff --git a/packages/ui/acp/README.md b/packages/ui/acp/README.md index d157a5158b..dd705e8d8d 100644 --- a/packages/ui/acp/README.md +++ b/packages/ui/acp/README.md @@ -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.) diff --git a/packages/ui/acp/src/index.ts b/packages/ui/acp/src/index.ts index ca6e31d008..c464ffd22b 100644 --- a/packages/ui/acp/src/index.ts +++ b/packages/ui/acp/src/index.ts @@ -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 = 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 } : {}, } } diff --git a/packages/ui/acp/tests/stream-update.spec.ts b/packages/ui/acp/tests/stream-update.spec.ts index 1112cc3808..2afa49e1d4 100644 --- a/packages/ui/acp/tests/stream-update.spec.ts +++ b/packages/ui/acp/tests/stream-update.spec.ts @@ -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 }) }) }) diff --git a/packages/ui/stdio-agent/README.md b/packages/ui/stdio-agent/README.md index 853abd52e2..5c37043b42 100644 --- a/packages/ui/stdio-agent/README.md +++ b/packages/ui/stdio-agent/README.md @@ -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 `''` 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` | diff --git a/packages/ui/stdio-agent/src/index.ts b/packages/ui/stdio-agent/src/index.ts index 65b38a0fd4..c3d6192a4b 100644 --- a/packages/ui/stdio-agent/src/index.ts +++ b/packages/ui/stdio-agent/src/index.ts @@ -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 } : {}, diff --git a/packages/ui/stdio-agent/tests/stdio-agent.spec.ts b/packages/ui/stdio-agent/tests/stdio-agent.spec.ts index 54853b7ca1..b5438caf77 100644 --- a/packages/ui/stdio-agent/tests/stdio-agent.spec.ts +++ b/packages/ui/stdio-agent/tests/stdio-agent.spec.ts @@ -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() })