diff --git a/CHANGELOG.md b/CHANGELOG.md index b2562648b..12379b87f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ ### Changed +- Parallel subagent batches now run at a fixed default of 8 children instead of a CPU-derived ceiling (`max(1, min(16, cores - 2))`), which squeezed small machines down to one or two children even though a child waits on a remote model, not on a local core. `[subagent] max_concurrency` (StepCode JSON: `subagent.maxConcurrency`) sets it per batch and may be higher or lower than the default; an invalid value is a settings warning and is ignored. `STEP_WORKFLOW_CONCURRENCY_LIMIT` no longer bounds subagent batches — it caps workflow `agent()` fan-out only — so a backend that limits simultaneous requests needs `[subagent] max_concurrency` as well. Batches still take slots from the process-wide admission window shared with workflow nodes, whose capacity is the smallest ceiling among the runs attached, so a concurrent workflow with a lower ceiling also lowers a batch and a 429 seen by either lowers both. See [Step configuration](docs/step-configuration.md#subagents). - `AgentLoopConfig.toolCallLeakRetries` is removed. The low-level loop no longer guesses a tool intent from markup in text; it records the structural classification (`classifyIncompleteAssistantTurn()`, now in `@step-harness/providers`) on every finished assistant message as `incomplete`, and hosts decide what to do about it. - Plan mode is now entered only by the user, with `/plan` or `--plan`; the `enter_plan_mode` tool is removed. The default system prompt no longer carries a `# Planning` section. Entering plan mode appends one hidden `` block with the plan method and plan path, and leaving it appends a short Default block, so the system prompt and tool block stay byte-identical across a plan cycle. Sessions recorded with an agent-initiated plan still restore. See [plan approval](docs/plan-approval.md#mode-announcements). - Subagents are no longer an automatic escalation. The `subagent` tool description now carries an explicit-request gate and a delegation method (critical path first, self-contained briefs, disjoint write sets, no redoing delegated work), replacing the system prompt's `# Delegation` section; `# Coordination primitives` only routes between authorized primitives. Session-standing `/ultracode on` also authorizes proactive delegation. See [orchestration lifecycle](docs/orchestration-lifecycle.md#subagent-delegation-authorization). @@ -25,7 +26,7 @@ - Stop appending Ultracode capability tutorials on every turn. Announce session-mode transitions after user input, reuse unchanged state, and restore necessary reminders when context is compacted or replaced. - Recognize `/exit` as an alias for `/quit`, including while the agent is running or compacting. - Recognize `search_files`, `find_files`, and `list_directory` when selecting default system-prompt file-exploration guidance. -- Report callback errors (including nested workflow errors raised before any agent starts), and queue ordinary subagent batches beyond their concurrency window in the same process-wide admission window as workflow nodes. Subagent batches also honor `STEP_WORKFLOW_CONCURRENCY_LIMIT` and hold each slot until the child process exits. +- Report callback errors (including nested workflow errors raised before any agent starts), and queue ordinary subagent batches beyond their own concurrency window in the same process-wide admission window as workflow nodes. A batch holds each slot until the child process exits. - Let a tool return `isError: true` while keeping its partial output and details, so failed subagent and workflow results reach the model as errors. Finalize a partial assistant message when a provider stream throws, ends without a terminal event, or fails during construction, instead of leaving the turn unfinished. - Classify provider failures structurally (HTTP status, provider error code, transport error) as rate limit, quota, auth, transient, or invalid request. Session retries, the separate 429 retry budget, and workflow congestion control use this classification instead of matching error text, so a 400 that mentions "429" is no longer retried and an exhausted quota returned as 429 is not treated as throttling. - Resume workflows by matching each call's prompt and options instead of the journal's sequence prefix, so a failed or reordered node no longer forces every later call to re-run. Once a live call writes files, runs a shell command, or uses an unrecognized tool, cached calls that read the environment re-run as well; cached results are re-validated against the call's schema, and calls with a runtime (safeParse) schema always run live. diff --git a/apps/cli/src/bootstrap/extensions.ts b/apps/cli/src/bootstrap/extensions.ts index 95d8fe77c..f19c787ef 100644 --- a/apps/cli/src/bootstrap/extensions.ts +++ b/apps/cli/src/bootstrap/extensions.ts @@ -62,6 +62,10 @@ export function createStepExtensionFactories(deps: StepExtensionFactoryDeps): In } : undefined; }, + maxConcurrencySettings: () => { + const settings = deps.stepSettings?.()?.getStepSettings(); + return settings ? { maxConcurrency: settings.subagentMaxConcurrency } : undefined; + }, }, }), createStepCronExtension({ telemetry: deps.telemetry }), diff --git a/docs/step-configuration.md b/docs/step-configuration.md index a99b4878a..d7182f488 100644 --- a/docs/step-configuration.md +++ b/docs/step-configuration.md @@ -120,22 +120,37 @@ Step screens pass the Step agent directory; generic TUI callers default to the system temporary directory. Rendering equivalence tests select the uncached renderer through a test-process argument. -### Background subagents +### Subagents -The `[subagent]` table controls background subagent lanes: +The `[subagent]` table controls subagent concurrency and background lanes: ```toml [subagent] background = true # false runs run_in_background calls in the foreground background_timeout_seconds = 3600 # default per-turn wall-clock limit of a lane +max_concurrency = 8 # children running at once across parallel batches ``` A trusted project's `config.toml` overrides the global values. `background_timeout_seconds` must be an integer from 60 to 7,200; any other value is reported as a settings warning and ignored, so the 3,600-second default -applies. A call's own `timeout_seconds` takes precedence. When -`STEPCODE_CONFIG_PATH` selects a StepCode JSON config, the same settings live -under `subagent.background` and `subagent.backgroundTimeoutSeconds`. +applies. A call's own `timeout_seconds` takes precedence. + +`max_concurrency` sets how many child processes run at once across parallel +subagent batches. It applies to foreground and print batches too, not only to +background lanes, and is a different limit from the at most eight running lanes +per session described in the [orchestration +lifecycle](orchestration-lifecycle.md#background-subagent-lanes). It defaults to +8 and must be a positive integer; there is no separate upper bound, so a +configured value may raise the limit above the default, and the batch itself is +still capped by `maxParallelTasks` (1,000 by default). Batches take slots from +the same process-wide admission window as workflow nodes, whose capacity is the +smallest ceiling among the runs currently attached: a concurrent workflow with a +lower ceiling also lowers a batch, and a 429 seen by either lowers both. The +value is read on every call, so a config reload takes effect without a restart. +When `STEPCODE_CONFIG_PATH` selects a StepCode JSON config, the same settings +live under `subagent.background` and `subagent.backgroundTimeoutSeconds`; +`max_concurrency` lives under `subagent.maxConcurrency`. `STEP_DISABLE_SUBAGENT_BACKGROUND=1` disables background lanes regardless of the file. Print and JSON runs never start lanes. See the [orchestration lifecycle](orchestration-lifecycle.md#background-subagent-lanes) for lane diff --git a/packages/coding-agent/docs/settings.md b/packages/coding-agent/docs/settings.md index 7e8b060a3..9667e182a 100644 --- a/packages/coding-agent/docs/settings.md +++ b/packages/coding-agent/docs/settings.md @@ -158,9 +158,9 @@ Temporary 429/concurrency errors retry the current model request, preserving com ### Workflow concurrency and failures -The harness does not pattern-match requests such as `开10个子agent` / `launch 10 subagents`; the model decides whether they call for the `subagent` tool's parallel mode or a workflow, and whether the message is a request at all. The `subagent` tool accepts batches larger than its concurrent window (up to 1000 by default) and queues the excess. Its children and workflow nodes take slots from the same process-wide admission window, so running both at once cannot exceed the host ceiling, and a 429 seen by either lowers it for both. +The harness does not pattern-match requests such as `开10个子agent` / `launch 10 subagents`; the model decides whether they call for the `subagent` tool's parallel mode or a workflow, and whether the message is a request at all. The `subagent` tool accepts batches larger than its own window (`[subagent] max_concurrency`, default 8, at most `maxParallelTasks` = 1,000 by default) and queues the excess. Its children and workflow nodes take slots from the same process-wide admission window, whose capacity is the smallest ceiling among the runs currently attached: a batch and a workflow running together are bound by the more conservative of the two, and a 429 seen by either lowers it for both. -The workflow ceiling is `max(1, min(16, CPU count - 2))`. Model-authored `maxConcurrency` can only lower this ceiling. Set `STEP_WORKFLOW_CONCURRENCY_LIMIT=8` to impose a lower trusted provider limit (for example, a backend allowing eight simultaneous requests). The effective ceiling is the minimum of the machine and configured limits; the final result reports requested, effective, and current concurrency. +The workflow ceiling is `max(1, min(16, CPU count - 2))`. Model-authored `maxConcurrency` can only lower this ceiling. Set `STEP_WORKFLOW_CONCURRENCY_LIMIT=8` to impose a lower trusted provider limit on workflow `agent()` calls (for example, a backend allowing eight simultaneous requests); it does not bound subagent batches, which use `[subagent] max_concurrency`. The effective workflow ceiling is the minimum of the machine and configured limits; the final result reports requested, effective, and current concurrency. Rate limits are recognized from the provider's structured classification (HTTP 429 or a rate-limit error code), never from text: error messages, agent output, and node labels that mention "429" or "rate limit" are not treated as congestion, and a 429 whose error code reports exhausted quota or billing (`insufficient_quota`, `GoUsageLimitError`, HTTP 402) is terminal. Temporary 429s reduce the admission cap by 25% (minimum one), once per congestion epoch. New nodes wait during backoff; running nodes keep their slots and retry their current request. Four successful responses allow a one-slot recovery when work is queued. Concurrent workflows share a process-wide host window, in addition to each run's lower limit. This conservatively covers all model overrides; it is not a cross-process/account-wide quota lock. diff --git a/packages/coding-agent/src/features/step-subagent.ts b/packages/coding-agent/src/features/step-subagent.ts index c2709d69b..264d79d78 100644 --- a/packages/coding-agent/src/features/step-subagent.ts +++ b/packages/coding-agent/src/features/step-subagent.ts @@ -20,9 +20,11 @@ import type { NewSessionOptions } from "../core/session-manager.ts"; import { resolveStepAgentDir, resolveStepConfigDir } from "../step/environment.ts"; import { isValidSubagentBackgroundTimeout, + isValidSubagentMaxConcurrency, SUBAGENT_BACKGROUND_TIMEOUT_DEFAULT_SECONDS, SUBAGENT_BACKGROUND_TIMEOUT_MAX_SECONDS, SUBAGENT_BACKGROUND_TIMEOUT_MIN_SECONDS, + SUBAGENT_MAX_CONCURRENCY_DEFAULT, } from "../step/settings-manager.ts"; import type { StepTelemetryReporter } from "../step/telemetry.ts"; import type { StepAgentConfig, StepAgentScope } from "./step-subagent-agents.ts"; @@ -49,7 +51,6 @@ import { } from "./subagent/rendering.ts"; import { createSubagentRpcSession } from "./subagent/rpc-adapter.ts"; import { buildSubagentToolDescription } from "./subagent/tool-description.ts"; -import { clampWorkflowConcurrency } from "./workflow/budget.ts"; // Lane-notification primitives moved to ./subagent/lane-events.ts; re-exported // here to keep this module's public surface stable. @@ -243,7 +244,11 @@ export interface StepSubagentExtensionOptions { includeBuiltinAgents?: boolean; /** Maximum batch size accepted in one parallel call; excess tasks queue instead of running. */ maxParallelTasks?: number; - /** Number of child processes allowed to run at once; tasks beyond this stay queued. */ + /** + * Fallback number of child processes allowed to run at once, used only when + * `[subagent] max_concurrency` is unset; a configured value overrides it. + * Must be a positive integer. + */ maxConcurrency?: number; /** Optional isolated git worktree implementation. */ worktreeManager?: StepWorktreeManager; @@ -256,6 +261,13 @@ export interface StepSubagentExtensionOptions { * `config.toml`), read on every call so a reload takes effect. */ backgroundSettings?: () => StepSubagentBackgroundSettings | undefined; + /** + * Host-supplied `[subagent] max_concurrency` (config.toml), read on every + * call so a reload takes effect. The settings boundary validates the value, + * so it is trusted here and overrides both the `maxConcurrency` option and + * the built-in default; it may exceed the default. + */ + maxConcurrencySettings?: () => StepSubagentConcurrencySettings | undefined; } export interface StepSubagentBackgroundSettings { @@ -265,6 +277,25 @@ export interface StepSubagentBackgroundSettings { timeoutSeconds?: number; } +export interface StepSubagentConcurrencySettings { + /** Default number of child processes a parallel batch runs at once. */ + maxConcurrency?: number; +} + +/** + * Resolve the parallel-batch ceiling: a validated configured value, then a + * validated host option, then the default. There is deliberately no CPU- or + * provider-derived ceiling — a child waits on a remote model, not on a local + * core — but an invalid value must never reach the shared admission window, + * where a non-positive or non-finite capacity wedges the process-wide pool + * instead of failing one batch. + */ +export function resolveSubagentConcurrency(configured?: number, option?: number): number { + if (isValidSubagentMaxConcurrency(configured)) return configured; + if (isValidSubagentMaxConcurrency(option)) return option; + return SUBAGENT_MAX_CONCURRENCY_DEFAULT; +} + // Keep the model-facing contract identical to Pi's subagent example. Step // adapts discovery, process invocation, storage, telemetry, and theme only. export const StepTaskItem = Type.Object( @@ -737,20 +768,25 @@ async function withSubagentListWidget( /** Construct the hidden inline extension used by the Step launcher. */ export function createStepSubagentExtension(options: StepSubagentExtensionOptions = {}): ExtensionFactory { + // The configured setting is validated where it is read; a host option is a + // programming error, so reject it here while the caller can still see it. + if (options.maxConcurrency !== undefined && !isValidSubagentMaxConcurrency(options.maxConcurrency)) { + throw new Error( + `Invalid Step subagent maxConcurrency option: ${String(options.maxConcurrency)} (expected a positive integer)`, + ); + } const resolved = { agentDir: path.resolve(options.agentDir ?? resolveStepAgentDir()), configDirName: options.configDirName?.trim() || resolveStepConfigDir() || CONFIG_DIR_NAME, includeBuiltinAgents: options.includeBuiltinAgents !== false, maxParallelTasks: Math.max(1, Math.min(MAX_PARALLEL_TASKS, options.maxParallelTasks ?? MAX_PARALLEL_TASKS)), - maxConcurrency: Math.max( - 1, - Math.min(clampWorkflowConcurrency(options.maxConcurrency), options.maxParallelTasks ?? MAX_PARALLEL_TASKS), - ), + maxConcurrency: resolveSubagentConcurrency(undefined, options.maxConcurrency), telemetry: options.telemetry, worktreeManager: options.worktreeManager ?? { allocate: allocateStepWorktree, }, runner: options.runner ?? runStepSubagentProcess, + maxConcurrencySettings: options.maxConcurrencySettings, }; return (pi: ExtensionAPI): void => { // The child process inherits this marker. It still gets Step's provider and diff --git a/packages/coding-agent/src/features/subagent/execute.ts b/packages/coding-agent/src/features/subagent/execute.ts index 9720e33ee..e0c793c5b 100644 --- a/packages/coding-agent/src/features/subagent/execute.ts +++ b/packages/coding-agent/src/features/subagent/execute.ts @@ -19,6 +19,7 @@ import { finalOutput, isFailed, makeToolResult, + resolveSubagentConcurrency, resultText, type StepChainItem, type StepSubagentDetails, @@ -41,11 +42,7 @@ import { type StepAgentScope, } from "../step-subagent-agents.ts"; import { acquireWorkflowAdmission } from "../workflow/admission.ts"; -import { - clampWorkflowConcurrency, - configuredWorkflowConcurrencyLimit, - type WorkflowSemaphore, -} from "../workflow/budget.ts"; +import type { WorkflowSemaphore } from "../workflow/budget.ts"; import { truncateText } from "./lane-events.ts"; import type { StepSubagentLaneRuntime } from "./lane-lifecycle.ts"; import { finalRoster, liveRoster } from "./projection.ts"; @@ -264,11 +261,12 @@ export type SubagentExecutionOptions = Required< StepSubagentExtensionOptions, "agentDir" | "configDirName" | "includeBuiltinAgents" | "maxParallelTasks" | "maxConcurrency" > -> & { - worktreeManager: StepWorktreeManager; - runner: StepSubagentRunner; - telemetry?: StepTelemetryReporter; -}; +> & + Pick & { + worktreeManager: StepWorktreeManager; + runner: StepSubagentRunner; + telemetry?: StepTelemetryReporter; + }; /** A validated, approved call: discovery ran, exactly one mode was selected, and project agents were confirmed. */ export interface PreparedSubagentRun { @@ -471,8 +469,10 @@ export async function runPreparedSubagent( return failed; } reportCreated(agent, index); - // A parallel batch shares the process-wide window with workflow nodes, so the - // two together never exceed the host ceiling. The task stays queued until admitted. + // A parallel batch shares the process-wide window with workflow nodes. The + // window's capacity is the smallest ceiling among the runs attached, so a batch + // and a workflow running together are bound by the more conservative of the two. + // The task stays queued until admitted. let releaseSlot: (() => void) | undefined; if (admission) { try { @@ -623,8 +623,14 @@ export async function runPreparedSubagent( // at 0s, including ones still queued behind maxConcurrency. emit(mode); if (parallel) { - // The configured provider limit caps subagent batches the same way it caps workflows. - const ceiling = Math.min(options.maxConcurrency, clampWorkflowConcurrency(configuredWorkflowConcurrencyLimit())); + // Plain default plus an optional configured value; no CPU- or provider-derived + // ceiling. Both inputs are validated here so an invalid host option cannot reach + // the shared window as a non-positive capacity. + const configured = options.maxConcurrencySettings?.()?.maxConcurrency; + const ceiling = Math.min( + resolveSubagentConcurrency(configured, options.maxConcurrency), + options.maxParallelTasks, + ); const lease = acquireWorkflowAdmission(ceiling); try { await mapWithConcurrency(parallel, ceiling, (task, index) => runOne(task, index, lease.semaphore)); diff --git a/packages/coding-agent/src/features/workflow/budget.ts b/packages/coding-agent/src/features/workflow/budget.ts index cf3f68cd0..98579719a 100644 --- a/packages/coding-agent/src/features/workflow/budget.ts +++ b/packages/coding-agent/src/features/workflow/budget.ts @@ -131,10 +131,15 @@ export class WorkflowSemaphore { } setCeiling(ceiling: number): void { - if (ceiling === this.ceiling) return; + // The pool's capacity is the smallest ceiling among attached members, so one + // caller can hand over a non-positive or non-finite value. That capacity admits + // nothing and would wedge the process-wide window instead of failing one run, so + // clamp it here as the last line of defense. + const next = Number.isFinite(ceiling) ? Math.max(1, Math.floor(ceiling)) : 1; + if (next === this.ceiling) return; const wasAtCeiling = this.limit === this.ceiling; - this.ceiling = ceiling; - this.limit = wasAtCeiling ? ceiling : Math.min(this.limit, ceiling); + this.ceiling = next; + this.limit = wasAtCeiling ? next : Math.min(this.limit, next); this.generation++; this.successes = 0; this.drain(); diff --git a/packages/coding-agent/src/step/settings-manager.ts b/packages/coding-agent/src/step/settings-manager.ts index 19d217aa4..ae09cb1f5 100644 --- a/packages/coding-agent/src/step/settings-manager.ts +++ b/packages/coding-agent/src/step/settings-manager.ts @@ -45,6 +45,8 @@ export interface StepSettings { subagentBackground?: boolean; /** `[subagent] background_timeout_seconds`: default per-turn wall-clock limit of a background lane. */ subagentBackgroundTimeoutSeconds?: number; + /** `[subagent] max_concurrency`: default number of child processes a parallel subagent batch runs at once. */ + subagentMaxConcurrency?: number; } type JsonObject = Record; @@ -65,8 +67,11 @@ export function isValidSubagentBackgroundTimeout(value: unknown): value is numbe } /** Message for a configured timeout that is ignored because it is out of bounds or not an integer. */ -export function invalidSubagentBackgroundTimeoutMessage(value: unknown): string { - return `[subagent] background_timeout_seconds must be an integer from ${SUBAGENT_BACKGROUND_TIMEOUT_MIN_SECONDS} to ${SUBAGENT_BACKGROUND_TIMEOUT_MAX_SECONDS}; ignoring ${JSON.stringify(value)} and using ${SUBAGENT_BACKGROUND_TIMEOUT_DEFAULT_SECONDS}`; +export function invalidSubagentBackgroundTimeoutMessage( + value: unknown, + keyPath = "[subagent] background_timeout_seconds", +): string { + return `${keyPath} must be an integer from ${SUBAGENT_BACKGROUND_TIMEOUT_MIN_SECONDS} to ${SUBAGENT_BACKGROUND_TIMEOUT_MAX_SECONDS}; ignoring ${JSON.stringify(value)} and using ${SUBAGENT_BACKGROUND_TIMEOUT_DEFAULT_SECONDS}`; } /** The raw configured timeout values, from the flat key and the `[subagent]` table. */ @@ -76,6 +81,32 @@ function rawSubagentTimeouts(value: JsonObject): unknown[] { ); } +/** Default number of child processes a parallel subagent batch runs at once. */ +export const SUBAGENT_MAX_CONCURRENCY_DEFAULT = 8; + +/** + * Whether `value` is an accepted `[subagent] max_concurrency`: a positive + * integer. There is deliberately no upper bound — raising the limit above the + * default is the point of the setting. + */ +export function isValidSubagentMaxConcurrency(value: unknown): value is number { + return typeof value === "number" && Number.isInteger(value) && value >= 1; +} + +/** Message for a configured concurrency that is ignored because it is not a positive integer. */ +export function invalidSubagentMaxConcurrencyMessage(value: unknown, keyPath = "[subagent] max_concurrency"): string { + // The effective fallback (option, then default) is resolved per call, so the + // load-time warning only states that the value is ignored. + return `${keyPath} must be a positive integer; ignoring ${JSON.stringify(value)}`; +} + +/** The raw configured concurrency values, from the flat key and the `[subagent]` table. */ +function rawSubagentConcurrencies(value: JsonObject): unknown[] { + return [value.subagentMaxConcurrency, asObject(value.subagent)?.max_concurrency].filter( + (candidate) => candidate !== undefined, + ); +} + export interface StepSettingsPaths { global: string; project: string; @@ -272,6 +303,10 @@ function normalizeSettings(value: JsonObject, resetLegacy = true): StepSettings isValidSubagentBackgroundTimeout, ); if (subagentTimeout !== undefined) result.subagentBackgroundTimeoutSeconds = subagentTimeout; + const subagentConcurrency = [value.subagentMaxConcurrency, subagent?.max_concurrency].find( + isValidSubagentMaxConcurrency, + ); + if (subagentConcurrency !== undefined) result.subagentMaxConcurrency = subagentConcurrency; if (resetLegacy) resetLegacyPermissionSettings(result, winningPresetCandidate, winningModeCandidate); return result; } @@ -299,6 +334,7 @@ const STEP_SETTING_ALIASES: Record = { feedbackEnabled: [], subagentBackground: [], subagentBackgroundTimeoutSeconds: [], + subagentMaxConcurrency: [], }; const STEP_SETTING_NESTED_ALIASES: Record = { @@ -329,6 +365,7 @@ const STEP_SETTING_NESTED_ALIASES: Record): Partial { `Invalid Step subagentBackgroundTimeoutSeconds setting: ${String(settings.subagentBackgroundTimeoutSeconds)} (expected an integer from ${SUBAGENT_BACKGROUND_TIMEOUT_MIN_SECONDS} to ${SUBAGENT_BACKGROUND_TIMEOUT_MAX_SECONDS})`, ); } + if ( + settings.subagentMaxConcurrency !== undefined && + !isValidSubagentMaxConcurrency(settings.subagentMaxConcurrency) + ) { + throw new Error( + `Invalid Step subagentMaxConcurrency setting: ${String(settings.subagentMaxConcurrency)} (expected a positive integer)`, + ); + } const normalized = { ...settings }; if (settings.permissionPreset !== undefined) { normalized.permissionPreset = getStepPermissionPreset(settings.permissionPreset)!.id; @@ -498,6 +543,10 @@ class StepSettingsStore { if (isValidSubagentBackgroundTimeout(timeout)) continue; this.errors.push({ scope, path, error: new Error(invalidSubagentBackgroundTimeoutMessage(timeout)) }); } + for (const concurrency of rawSubagentConcurrencies(value)) { + if (isValidSubagentMaxConcurrency(concurrency)) continue; + this.errors.push({ scope, path, error: new Error(invalidSubagentMaxConcurrencyMessage(concurrency)) }); + } return value; } diff --git a/packages/coding-agent/src/step/stepcode-config.ts b/packages/coding-agent/src/step/stepcode-config.ts index 639bbbbde..65e2cdce6 100644 --- a/packages/coding-agent/src/step/stepcode-config.ts +++ b/packages/coding-agent/src/step/stepcode-config.ts @@ -7,6 +7,7 @@ import type { AnthropicMessagesCompat, Api } from "@step-harness/providers"; import lockfile from "proper-lockfile"; import type { ExtensionAPI, InlineExtension, ProviderConfig, ProviderModelConfig } from "../core/extensions/types.ts"; import { normalizeProviderBaseUrl } from "../core/provider-base-url.ts"; +import type { SettingsError } from "../core/settings-manager.ts"; import { stripBom } from "../utils/text.ts"; import { getStepPermissionPreset, @@ -16,7 +17,10 @@ import { type StepPermissionPresetId, } from "./permissions.ts"; import { + invalidSubagentBackgroundTimeoutMessage, + invalidSubagentMaxConcurrencyMessage, isValidSubagentBackgroundTimeout, + isValidSubagentMaxConcurrency, resetLegacyPermissionSettings, type StepSettings, type StepSettingsManager, @@ -152,6 +156,7 @@ export function decorateStepCodeSettingsManager( "getStepAutoResume", "setStepAutoResume", "reload", + "drainErrors", ]); return new Proxy(manager, { get(target, property) { @@ -174,6 +179,8 @@ class StepCodeSettingsAuthority { private defaultProvider: string | undefined; private defaultModel: string | undefined; private settings: StepSettings; + /** Load-time diagnostics for values this authority ignored, drained by the host. */ + private readonly loadErrors: SettingsError[] = []; constructor(manager: StepSettingsManager, config: StepCodeConfig) { this.manager = manager; @@ -182,7 +189,12 @@ class StepCodeSettingsAuthority { const storedDefault = findRawActiveModel(root); this.defaultProvider = storedDefault?.providerId ?? config.defaultProvider; this.defaultModel = storedDefault?.modelId ?? config.defaultModel; - this.settings = readStepCodeSettings(root); + this.settings = readStepCodeSettings(root, (error) => this.reportLoadError(error)); + } + + /** Record an ignored value with the config path so drainErrors can warn about it. */ + private reportLoadError(error: Error): void { + this.loadErrors.push({ scope: "global", path: this.path, error }); } getDefaultProvider(): string | undefined { @@ -284,7 +296,15 @@ class StepCodeSettingsAuthority { const config = await loadStepCodeConfig({ ...process.env, STEPCODE_CONFIG_PATH: this.path }); this.defaultProvider = config?.defaultProvider; this.defaultModel = config?.defaultModel; - this.settings = readStepCodeSettings(readConfigRoot(this.path)); + // A reload replaces the whole snapshot: diagnostics from the previous file + // must not outlive a value the user has already fixed. + this.loadErrors.length = 0; + this.settings = readStepCodeSettings(readConfigRoot(this.path), (error) => this.reportLoadError(error)); + } + + /** Drain this authority's load diagnostics together with the wrapped manager's. */ + drainErrors(): Array[number]> { + return [...this.manager.drainErrors(), ...this.loadErrors.splice(0)]; } private updateConfig(update: (root: Record) => void): void { @@ -334,7 +354,7 @@ function acquireConfigLock(path: string): () => void { throw (lastError as Error) ?? new Error(`Failed to lock StepCode config ${path}`); } -function readStepCodeSettings(root: JsonObject): StepSettings { +function readStepCodeSettings(root: JsonObject, report?: (error: Error) => void): StepSettings { const approval = asObject(asObject(root.tools)?.approval) ?? asObject(root.approval); const preset = getStepPermissionPreset(readString(approval?.preset))?.id; const mode = normalizeStepPermissionMode(readString(approval?.mode)); @@ -348,6 +368,15 @@ function readStepCodeSettings(root: JsonObject): StepSettings { const subagent = asObject(root.subagent); const subagentBackground = readBoolean(subagent?.background); const subagentTimeout = subagent?.backgroundTimeoutSeconds; + if (subagentTimeout !== undefined && !isValidSubagentBackgroundTimeout(subagentTimeout)) { + report?.( + new Error(invalidSubagentBackgroundTimeoutMessage(subagentTimeout, "subagent.backgroundTimeoutSeconds")), + ); + } + const subagentConcurrency = subagent?.maxConcurrency; + if (subagentConcurrency !== undefined && !isValidSubagentMaxConcurrency(subagentConcurrency)) { + report?.(new Error(invalidSubagentMaxConcurrencyMessage(subagentConcurrency, "subagent.maxConcurrency"))); + } const settings: StepSettings = { ...(preset ? { permissionPreset: preset } : {}), ...(mode ? { approvalMode: mode } : {}), @@ -359,6 +388,7 @@ function readStepCodeSettings(root: JsonObject): StepSettings { ...(isValidSubagentBackgroundTimeout(subagentTimeout) ? { subagentBackgroundTimeoutSeconds: subagentTimeout } : {}), + ...(isValidSubagentMaxConcurrency(subagentConcurrency) ? { subagentMaxConcurrency: subagentConcurrency } : {}), }; resetLegacyPermissionSettings(settings, readString(approval?.preset), readString(approval?.mode)); return settings; @@ -380,10 +410,15 @@ function writeStepCodeSettings(root: Record, patch: Partial): void { `Invalid Step subagentBackgroundTimeoutSeconds setting: ${String(settings.subagentBackgroundTimeoutSeconds)}`, ); } + if ( + settings.subagentMaxConcurrency !== undefined && + !isValidSubagentMaxConcurrency(settings.subagentMaxConcurrency) + ) { + throw new Error(`Invalid Step subagentMaxConcurrency setting: ${String(settings.subagentMaxConcurrency)}`); + } } function findRawModelHandle(root: JsonObject, providerId: string, modelId: string): string | undefined { diff --git a/packages/coding-agent/test/step-settings-manager.test.ts b/packages/coding-agent/test/step-settings-manager.test.ts index 2e7734a12..7c3fd5e2d 100644 --- a/packages/coding-agent/test/step-settings-manager.test.ts +++ b/packages/coding-agent/test/step-settings-manager.test.ts @@ -297,6 +297,29 @@ describe("Step settings manager decorator", () => { ); }); + it("reads the [subagent] max_concurrency setting and ignores invalid values", () => { + const fixture = makeRoot(); + writeFileSync(fixture.paths.global, JSON.stringify({ subagent: { max_concurrency: 12 } })); + writeFileSync(fixture.paths.project, JSON.stringify({ subagent: { max_concurrency: -5 } })); + const { manager } = makeManager(fixture); + + expect(manager.getStepSettings()).toEqual({ subagentMaxConcurrency: 12 }); + expect(() => manager.setStepSettings({ subagentMaxConcurrency: 0 })).toThrow(/subagentMaxConcurrency/u); + // The ignored project value must surface as a warning, not vanish silently. + const errors = manager.drainErrors(); + expect(errors.some((entry) => entry.scope === "project" && /max_concurrency/u.test(entry.error.message))).toBe( + true, + ); + }); + + it("reads the flat subagentMaxConcurrency key", () => { + const fixture = makeRoot(); + writeFileSync(fixture.paths.global, JSON.stringify({ subagentMaxConcurrency: 16 })); + const { manager } = makeManager(fixture); + + expect(manager.getStepSettings()).toEqual({ subagentMaxConcurrency: 16 }); + }); + it("lets a project feedback alias override the global canonical setting", () => { const fixture = makeRoot(); writeFileSync(fixture.paths.global, JSON.stringify({ feedbackEnabled: true })); diff --git a/packages/coding-agent/test/stepcode-config.test.ts b/packages/coding-agent/test/stepcode-config.test.ts index 2b039d4e6..a14a1f536 100644 --- a/packages/coding-agent/test/stepcode-config.test.ts +++ b/packages/coding-agent/test/stepcode-config.test.ts @@ -205,7 +205,89 @@ describe("StepCode config compatibility", () => { ).toThrow(/subagentBackgroundTimeoutSeconds/u); }); - test("ignores an out-of-range subagent timeout in the injected config", async () => { + test("reads and writes the subagent max_concurrency in the injected config", async () => { + const root = await mkdtemp(join(tmpdir(), "stepcode-config-subagent-concurrency-")); + roots.push(root); + const configPath = join(root, "stepcode", "config.json"); + const agentDir = join(root, "stepcode", "agent"); + const projectDir = join(root, "project"); + await mkdir(agentDir, { recursive: true }); + await mkdir(projectDir, { recursive: true }); + await writeFile( + configPath, + JSON.stringify({ + providers: { p: { api: "anthropic-messages", baseUrl: "https://proxy.example/v1", models: [{ id: "m" }] } }, + activeModel: "m", + subagent: { maxConcurrency: 12 }, + }), + ); + const config = await loadStepCodeConfig({ STEPCODE_CONFIG_PATH: configPath }, root); + const manager = decorateStepCodeSettingsManager(createStepSettingsManager(projectDir, agentDir), config!); + expect(manager.getStepSettings()).toMatchObject({ subagentMaxConcurrency: 12 }); + + manager.setEffectiveStepSettings({ ...manager.getStepSettings(), subagentMaxConcurrency: 16 }); + const saved = JSON.parse(readFileSync(configPath, "utf8")) as { subagent: Record }; + expect(saved.subagent).toEqual({ maxConcurrency: 16 }); + expect(() => + manager.setEffectiveStepSettings({ ...manager.getStepSettings(), subagentMaxConcurrency: 0 }), + ).toThrow(/subagentMaxConcurrency/u); + }); + + test("ignores an invalid subagent max_concurrency in the injected config", async () => { + const root = await mkdtemp(join(tmpdir(), "stepcode-config-subagent-concurrency-bad-")); + roots.push(root); + const configPath = join(root, "config.json"); + const agentDir = join(root, "agent"); + await mkdir(agentDir, { recursive: true }); + await writeFile( + configPath, + JSON.stringify({ + providers: { p: { api: "anthropic-messages", baseUrl: "https://proxy.example/v1", models: [{ id: "m" }] } }, + activeModel: "m", + subagent: { maxConcurrency: 2.5 }, + }), + ); + const config = await loadStepCodeConfig({ STEPCODE_CONFIG_PATH: configPath }, root); + const manager = decorateStepCodeSettingsManager(createStepSettingsManager(root, agentDir), config!); + expect(manager.getStepSettings().subagentMaxConcurrency).toBeUndefined(); + // The ignored value must surface as a settings warning naming the JSON key, + // not fail silently and not point the user at the TOML key. + const errors = manager.drainErrors(); + expect(errors.some((entry) => /subagent\.maxConcurrency/u.test(entry.error.message))).toBe(true); + expect(errors.some((entry) => entry.path === configPath)).toBe(true); + expect(manager.drainErrors()).toEqual([]); + }); + + test("drops stale max_concurrency warnings on reload", async () => { + const root = await mkdtemp(join(tmpdir(), "stepcode-config-subagent-concurrency-reload-")); + roots.push(root); + const configPath = join(root, "config.json"); + const agentDir = join(root, "agent"); + await mkdir(agentDir, { recursive: true }); + const write = (maxConcurrency: number) => + writeFile( + configPath, + JSON.stringify({ + providers: { + p: { api: "anthropic-messages", baseUrl: "https://proxy.example/v1", models: [{ id: "m" }] }, + }, + activeModel: "m", + subagent: { maxConcurrency }, + }), + ); + await write(2.5); + const config = await loadStepCodeConfig({ STEPCODE_CONFIG_PATH: configPath }, root); + const manager = decorateStepCodeSettingsManager(createStepSettingsManager(root, agentDir), config!); + expect(manager.drainErrors()).toHaveLength(1); + + await write(12); + await manager.reload(); + // The fixed value produces no new diagnostics, and the old one is gone. + expect(manager.drainErrors()).toEqual([]); + expect(manager.getStepSettings().subagentMaxConcurrency).toBe(12); + }); + + test("warns about an invalid subagent timeout in the injected config", async () => { const root = await mkdtemp(join(tmpdir(), "stepcode-config-subagent-bad-")); roots.push(root); const configPath = join(root, "config.json"); @@ -222,6 +304,10 @@ describe("StepCode config compatibility", () => { const config = await loadStepCodeConfig({ STEPCODE_CONFIG_PATH: configPath }, root); const manager = decorateStepCodeSettingsManager(createStepSettingsManager(root, agentDir), config!); expect(manager.getStepSettings().subagentBackgroundTimeoutSeconds).toBeUndefined(); + // The two adjacent keys behave the same way: an ignored value is a warning. + expect( + manager.drainErrors().some((entry) => /subagent\.backgroundTimeoutSeconds/u.test(entry.error.message)), + ).toBe(true); }); test("keeps StepCode model and permission reads and writes in the injected config", async () => { diff --git a/packages/coding-agent/test/subagent-chain-result.test.ts b/packages/coding-agent/test/subagent-chain-result.test.ts index 1099d3328..c4643e065 100644 --- a/packages/coding-agent/test/subagent-chain-result.test.ts +++ b/packages/coding-agent/test/subagent-chain-result.test.ts @@ -284,18 +284,16 @@ test("a 429 inside a parallel subagent lowers the shared window", async () => { workflow.release(); }); -test("parallel subagents honor the configured provider concurrency limit", async () => { +test("parallel subagents ignore the workflow provider concurrency limit", async () => { + // A subagent batch's ceiling is a plain default plus [subagent] max_concurrency; + // STEP_WORKFLOW_CONCURRENCY_LIMIT only caps workflow agent() fan-out. vi.stubEnv("STEP_WORKFLOW_CONCURRENCY_LIMIT", "1"); const tasks = ["a", "b", "c"].map((task) => ({ agent: "explore", task })); const { result, peakActive } = await run({ tasks } as SubagentParams, () => ({ text: "ok" }), { runnerDelayMs: 10, }); expect(contentText(result)).toContain("Parallel: 3/3 succeeded"); - expect(peakActive).toBe(1); - - vi.unstubAllEnvs(); - const unlimited = await run({ tasks } as SubagentParams, () => ({ text: "ok" }), { runnerDelayMs: 10 }); - expect(unlimited.peakActive).toBeGreaterThan(1); + expect(peakActive).toBeGreaterThan(1); }); test("admitted parallel subagents ask the runner to hold their slot until the child exits", async () => { diff --git a/packages/coding-agent/test/subagent-concurrency.test.ts b/packages/coding-agent/test/subagent-concurrency.test.ts new file mode 100644 index 000000000..0b4645911 --- /dev/null +++ b/packages/coding-agent/test/subagent-concurrency.test.ts @@ -0,0 +1,193 @@ +import os from "node:os"; +import type { AgentToolResult } from "@step-harness/agent-core"; +import { afterEach, beforeEach, expect, test, vi } from "vitest"; +import type { ExtensionContext } from "../src/core/extensions/types.ts"; +import type { + StepSubagentDetails, + StepSubagentExtensionOptions, + StepSubagentRunResult, + SubagentParams, +} from "../src/features/step-subagent.ts"; +import { createStepSubagentExtension, resolveSubagentConcurrency } from "../src/features/step-subagent.ts"; +import { executeSubagent } from "../src/features/subagent/execute.ts"; +import { resetWorkflowAdmission } from "../src/features/workflow/admission.ts"; + +beforeEach(() => { + // A 2-CPU host: the old CPU-derived default clamped parallel batches to + // max(1, min(16, 2 - 2)) = 1 child, so these tests prove the batch now runs + // at the plain default and any configured value regardless of host CPUs. + vi.spyOn(os, "cpus").mockReturnValue(Array(2).fill(os.cpus()[0]!)); + vi.stubEnv("STEP_WORKFLOW_CONCURRENCY_LIMIT", ""); +}); + +afterEach(() => { + vi.restoreAllMocks(); + vi.unstubAllEnvs(); + resetWorkflowAdmission(); +}); + +function okResult(text: string): StepSubagentRunResult { + return { + messages: [ + { + role: "assistant", + content: [{ type: "text", text }], + api: "anthropic-messages", + provider: "step", + model: "step-3.7-flash", + usage: { + input: 1, + output: 2, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 3, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "stop", + timestamp: Date.now(), + }, + ], + stderr: "", + exitCode: 0, + stopReason: "stop", + usage: { input: 1, output: 2, cost: 0, contextTokens: 3, turns: 1 }, + startedAt: Date.now(), + updatedAt: Date.now(), + } as unknown as StepSubagentRunResult; +} + +const ctx = { + mode: "print", + hasUI: false, + cwd: "/tmp", + model: undefined, + thinkingLevel: "high", + isProjectTrusted: () => true, +} as unknown as ExtensionContext; + +function baseOptions(overrides: Partial = {}): StepSubagentExtensionOptions { + return { + agentDir: "/tmp/step-agent-test", + configDirName: ".step", + includeBuiltinAgents: true, + ...overrides, + }; +} + +/** + * Run a parallel batch of `tasks` tasks and return the observed peak number of + * concurrently active children. Every runner blocks until the expected peak is + * reached (or a timeout fires), so the peak is observed deterministically + * instead of depending on runner timing. + */ +async function peakConcurrency(tasks: number, expected: number, options: StepSubagentExtensionOptions) { + let active = 0; + let peak = 0; + let release!: () => void; + const saturated = new Promise((resolve) => { + release = resolve; + }); + const result: AgentToolResult = await executeSubagent( + { + tasks: Array.from({ length: tasks }, (_unused, index) => ({ agent: "general", task: `task-${index}` })), + } as SubagentParams, + undefined, + () => {}, + ctx, + { + ...baseOptions(options), + // Keep a caller-provided batch cap; the default must not swallow it. + maxParallelTasks: options.maxParallelTasks ?? Math.max(1_000, tasks), + worktreeManager: { allocate: async () => Promise.reject(new Error("no worktrees in this test")) }, + runner: async (input) => { + active += 1; + peak = Math.max(peak, active); + if (active >= expected) release(); + await Promise.race([saturated, new Promise((resolve) => setTimeout(resolve, 250))]); + active -= 1; + return okResult(input.task); + }, + } as Parameters[4], + ); + return { peak, result }; +} + +test("the resolver defaults to 8 and prefers a valid configured value", () => { + expect(resolveSubagentConcurrency(undefined, undefined)).toBe(8); + expect(resolveSubagentConcurrency(undefined, 3)).toBe(3); + // A configured value wins over the option and may exceed the default. + expect(resolveSubagentConcurrency(12, 3)).toBe(12); + expect(resolveSubagentConcurrency(64, undefined)).toBe(64); +}); + +test("the resolver never returns an invalid ceiling", () => { + // A non-positive or non-finite capacity would wedge the process-wide admission + // window, so an invalid input falls through to the next valid source. + expect(resolveSubagentConcurrency(0, 3)).toBe(3); + expect(resolveSubagentConcurrency(-4, undefined)).toBe(8); + expect(resolveSubagentConcurrency(2.5, undefined)).toBe(8); + expect(resolveSubagentConcurrency(Number.NaN, 5)).toBe(5); + expect(resolveSubagentConcurrency(undefined, 0)).toBe(8); + expect(resolveSubagentConcurrency(undefined, Number.NaN)).toBe(8); + expect(resolveSubagentConcurrency(undefined, 2.5)).toBe(8); +}); + +test("an invalid host option is rejected when the extension is created", () => { + // The setting is validated where it is read, but a host option is a programmer + // error: reject it loudly instead of clamping it into an unexpected ceiling. + for (const invalid of [0, -1, 2.5, Number.NaN]) { + expect(() => createStepSubagentExtension({ agentDir: "/tmp/step-agent-test", maxConcurrency: invalid })).toThrow( + /maxConcurrency/u, + ); + } + expect(() => createStepSubagentExtension({ agentDir: "/tmp/step-agent-test", maxConcurrency: 12 })).not.toThrow(); +}); + +test("an invalid host option cannot deadlock a batch run through executeSubagent", async () => { + // Embedded hosts can bypass the factory; the execution path still has to fall + // back to a usable ceiling rather than queue every task behind a zero capacity. + for (const invalid of [0, -1, Number.NaN]) { + const { peak, result } = await peakConcurrency(8, 8, { maxConcurrency: invalid }); + expect(peak).toBe(8); + expect(result.details?.results.every((record) => record.status === "completed")).toBe(true); + } +}); + +test("a parallel batch runs at the default of 8 regardless of host CPUs", async () => { + const { peak, result } = await peakConcurrency(8, 8, {}); + expect(peak).toBe(8); + expect(result.details?.results.every((record) => record.status === "completed")).toBe(true); +}); + +test("a configured max_concurrency above the default raises the ceiling", async () => { + const { peak } = await peakConcurrency(12, 12, { + maxConcurrencySettings: () => ({ maxConcurrency: 12 }), + }); + expect(peak).toBe(12); +}); + +test("the workflow provider-limit env var no longer caps a subagent batch", async () => { + vi.stubEnv("STEP_WORKFLOW_CONCURRENCY_LIMIT", "2"); + const { peak } = await peakConcurrency(8, 8, {}); + expect(peak).toBe(8); +}); + +test("a configured value overrides the host option", async () => { + const { peak } = await peakConcurrency(6, 6, { + maxConcurrency: 2, + maxConcurrencySettings: () => ({ maxConcurrency: 6 }), + }); + expect(peak).toBe(6); +}); + +test("maxParallelTasks still rejects an oversized batch before any child starts", async () => { + const { peak, result } = await peakConcurrency(5, 5, { maxParallelTasks: 4 }); + expect(peak).toBe(0); + const text = (result.content ?? []).map((block) => (block.type === "text" ? block.text : "")).join(""); + expect(text).toContain("Too many parallel tasks (5); maximum is 4"); +}); + +test("tasks beyond the ceiling queue instead of running", async () => { + const { peak } = await peakConcurrency(5, 2, { maxConcurrency: 2 }); + expect(peak).toBe(2); +}); diff --git a/packages/coding-agent/test/workflow-runtime.test.ts b/packages/coding-agent/test/workflow-runtime.test.ts index 851eaec8b..b67e8a1ae 100644 --- a/packages/coding-agent/test/workflow-runtime.test.ts +++ b/packages/coding-agent/test/workflow-runtime.test.ts @@ -129,6 +129,21 @@ test("optional provider concurrency ignores blank and invalid environment values expect(configuredWorkflowConcurrencyLimit(" 8 ")).toBe(8); }); +test("a non-positive shared ceiling cannot wedge the admission window", async () => { + // The pool's capacity is the smallest ceiling among attached members, so a bad + // value from one caller must not become a capacity that admits nothing. + const semaphore = new WorkflowSemaphore(4); + semaphore.setCeiling(0); + expect(semaphore.max).toBe(1); + const release = await semaphore.acquire(); + release(); + semaphore.setCeiling(Number.NaN); + expect(semaphore.max).toBe(1); + const releaseAgain = await semaphore.acquire(); + releaseAgain(); + semaphore.dispose(); +}); + test("a long server Retry-After cannot freeze the shared admission window", async () => { // A node's own retry honors Retry-After verbatim, but the admission window is // process-wide, so an hour-long (or far-future-date) cooldown must not stall