From 3e24f26a36725db907f7512e819eaca4b20be9a0 Mon Sep 17 00:00:00 2001 From: Greg Pstrucha <875316+gricha@users.noreply.github.com> Date: Mon, 24 Aug 2026 08:31:34 -0700 Subject: [PATCH 1/2] perf(runner): Parallelize hunk analysis Move hunk analysis onto one shared FIFO work queue so large multi-hunk files can use the configured concurrency capacity. Preserve deterministic report order while applying one global limit across entry points. Co-Authored-By: GPT-5.6 --- .../docs/src/content/docs/architecture.mdx | 7 +- packages/docs/src/content/docs/cli/run.mdx | 2 +- .../docs/src/content/docs/config/runner.mdx | 2 +- .../docs/src/content/docs/github/workflow.mdx | 2 +- .../src/action/triggers/executor.test.ts | 4 +- .../warden/src/action/triggers/executor.ts | 11 +- .../__fixtures__/schedule/warden.toml | 3 + .../src/action/workflow/pr-workflow.test.ts | 20 +- .../warden/src/action/workflow/pr-workflow.ts | 8 +- .../src/action/workflow/schedule.test.ts | 6 +- .../warden/src/action/workflow/schedule.ts | 6 + packages/warden/src/cli/args.ts | 2 +- packages/warden/src/cli/help.ts | 2 +- packages/warden/src/cli/main.ts | 2 +- packages/warden/src/cli/output/ink-runner.tsx | 15 +- packages/warden/src/cli/output/tasks.test.ts | 132 ++++++++----- packages/warden/src/cli/output/tasks.ts | 91 ++++----- packages/warden/src/config/schema.ts | 4 +- packages/warden/src/sdk/analyze.test.ts | 142 ++++++++++++++ packages/warden/src/sdk/analyze.ts | 175 ++++++++++-------- packages/warden/src/sdk/types.ts | 10 +- packages/warden/src/utils/async.test.ts | 137 +++++++------- packages/warden/src/utils/async.ts | 109 ++++++++--- packages/warden/src/utils/index.ts | 2 +- skills/warden/references/cli-reference.md | 2 +- 25 files changed, 577 insertions(+), 319 deletions(-) diff --git a/packages/docs/src/content/docs/architecture.mdx b/packages/docs/src/content/docs/architecture.mdx index 40f5f683b..dbee2d5f9 100644 --- a/packages/docs/src/content/docs/architecture.mdx +++ b/packages/docs/src/content/docs/architecture.mdx @@ -72,7 +72,7 @@ A matched trigger becomes a skill task. Each task contains: - model, runtime, max turns, chunking, and verification options - output thresholds for failure and reporting -Warden launches matched skills in parallel. A shared semaphore gates file-level +Warden launches matched skills in parallel. A shared queue dispatches hunk analysis so multiple skills can be active while total model concurrency stays bounded. @@ -88,8 +88,9 @@ Before a model sees code, Warden prepares the diff: 4. Expand each hunk with surrounding file context. 5. Group hunks by file. -The unit of main analysis is a hunk with context. Files run in parallel when -allowed by the runner, while hunks inside a file run in order. +The unit of main analysis is a hunk with context. Hunks are independent queue +items, so workers may analyze multiple hunks from the same file concurrently. +Reports preserve source order even when hunks finish out of order. See [Chunking](/config/chunking/) for file pattern modes and coalescing settings. diff --git a/packages/docs/src/content/docs/cli/run.mdx b/packages/docs/src/content/docs/cli/run.mdx index ef46383dd..55562a11f 100644 --- a/packages/docs/src/content/docs/cli/run.mdx +++ b/packages/docs/src/content/docs/cli/run.mdx @@ -34,7 +34,7 @@ The bare `warden` command is an alias for this command. | `--report-on ` | Only show findings at or above this severity. | | `--min-confidence ` | Only show findings at or above this confidence. | | `--fix` | Automatically apply all suggested fixes. | -| `--parallel ` | Max concurrent file analyses across running skills. | +| `--parallel ` | Max concurrent hunk analyses across running skills. | | `-x, --fail-fast` | Stop after the first finding. | | `--staged` | Analyze only staged changes. | | `--git` | Force ambiguous targets to be treated as git refs. | diff --git a/packages/docs/src/content/docs/config/runner.mdx b/packages/docs/src/content/docs/config/runner.mdx index edcc974eb..418f66cdb 100644 --- a/packages/docs/src/content/docs/config/runner.mdx +++ b/packages/docs/src/content/docs/config/runner.mdx @@ -17,7 +17,7 @@ concurrency = 1 concurrency number -
Maximum concurrent file analyses across running CLI skills. In GitHub Actions, also limits matched trigger dispatch.
+
Maximum concurrent hunk analyses across running CLI skills. In GitHub Actions, it also limits matched trigger dispatch.
diff --git a/packages/docs/src/content/docs/github/workflow.mdx b/packages/docs/src/content/docs/github/workflow.mdx index ba4103a6c..2f73c7670 100644 --- a/packages/docs/src/content/docs/github/workflow.mdx +++ b/packages/docs/src/content/docs/github/workflow.mdx @@ -264,6 +264,6 @@ that do not actually run for the current event complete as neutral. parallel number -
Maximum concurrent matched trigger executions and file analyses unless runner.concurrency is set. Default: 5.
+
Maximum concurrent matched trigger executions and hunk analyses unless runner.concurrency is set. Default: 5.
diff --git a/packages/warden/src/action/triggers/executor.test.ts b/packages/warden/src/action/triggers/executor.test.ts index 714596c6f..d0e64cae0 100644 --- a/packages/warden/src/action/triggers/executor.test.ts +++ b/packages/warden/src/action/triggers/executor.test.ts @@ -232,7 +232,6 @@ describe('executeTrigger', () => { auxiliaryMaxRetries: 9, }), }), - expect.any(Number), expect.anything(), undefined ); @@ -252,7 +251,6 @@ describe('executeTrigger', () => { expect.objectContaining({ runnerOptions: expect.objectContaining({ historicalEvidence }), }), - expect.any(Number), expect.anything(), undefined ); @@ -611,7 +609,7 @@ describe('executeTrigger', () => { reason: 'not real', }; - vi.mocked(runSkillTask).mockImplementation(async (_taskOptions, _fileConcurrency, callbacks) => { + vi.mocked(runSkillTask).mockImplementation(async (_taskOptions, callbacks) => { callbacks.onFindingProcessing?.('test-trigger', event); return { name: 'test-trigger', report: mockReport }; }); diff --git a/packages/warden/src/action/triggers/executor.ts b/packages/warden/src/action/triggers/executor.ts index 0ded4aa15..20ae18e22 100644 --- a/packages/warden/src/action/triggers/executor.ts +++ b/packages/warden/src/action/triggers/executor.ts @@ -18,10 +18,10 @@ import { runSkillTask, createDefaultCallbacks } from '../../cli/output/tasks.js' import type { SkillTaskOptions } from '../../cli/output/tasks.js'; import { renderSkillReport } from '../../output/renderer.js'; import { logGroup, logGroupEnd } from '../workflow/base.js'; -import { DEFAULT_FILE_CONCURRENCY, type AnalysisChunkingConfig } from '../../sdk/types.js'; +import type { AnalysisChunkingConfig } from '../../sdk/types.js'; import type { FindingProcessingEvent } from '../../sdk/types.js'; import { SkillRunnerError } from '../../sdk/errors.js'; -import type { Semaphore } from '../../utils/index.js'; +import type { AsyncWorkQueue } from '../../utils/index.js'; import { Verbosity } from '../../cli/output/verbosity.js'; import type { ProviderFailureCircuitBreaker } from '../../sdk/circuit-breaker.js'; import { assertValidPiModelSelectors } from '../../sdk/runtimes/model-selectors.js'; @@ -100,8 +100,8 @@ export interface TriggerExecutorDeps { globalRequestChanges?: boolean; /** Global fail-check from action inputs (trigger-specific takes precedence) */ globalFailCheck?: boolean; - /** Global semaphore for limiting concurrent file analyses across triggers */ - semaphore?: Semaphore; + /** Global queue for limiting concurrent hunk analyses across triggers. */ + analysisQueue?: AsyncWorkQueue; /** Shared controller for stopping the whole action run */ abortController?: AbortController; /** Shared circuit breaker for auth/provider failures */ @@ -237,8 +237,7 @@ export async function executeTrigger( defaultCallbacks.onFindingProcessing?.(skillName, event); }, }; - const fileConcurrency = deps.semaphore ? Number.MAX_SAFE_INTEGER : DEFAULT_FILE_CONCURRENCY; - const result = await runSkillTask(taskOptions, fileConcurrency, callbacks, deps.semaphore); + const result = await runSkillTask(taskOptions, callbacks, deps.analysisQueue); const report = result.report; if (!report) { diff --git a/packages/warden/src/action/workflow/__fixtures__/schedule/warden.toml b/packages/warden/src/action/workflow/__fixtures__/schedule/warden.toml index 9a1398536..d943832d5 100644 --- a/packages/warden/src/action/workflow/__fixtures__/schedule/warden.toml +++ b/packages/warden/src/action/workflow/__fixtures__/schedule/warden.toml @@ -1,5 +1,8 @@ version = 1 +[runner] +concurrency = 2 + [defaults] auxiliaryMaxRetries = 3 diff --git a/packages/warden/src/action/workflow/pr-workflow.test.ts b/packages/warden/src/action/workflow/pr-workflow.test.ts index 1483f6683..648c5f4ec 100644 --- a/packages/warden/src/action/workflow/pr-workflow.test.ts +++ b/packages/warden/src/action/workflow/pr-workflow.test.ts @@ -129,7 +129,7 @@ import { } from './base.js'; import { runPRWorkflow } from './pr-workflow.js'; import { clearSkillsCache } from '../../skills/loader.js'; -import { Semaphore } from '../../utils/index.js'; +import { AsyncWorkQueue } from '../../utils/index.js'; import { buildFindingsOutput } from '../../reporting/output.js'; // Type the mocks @@ -1621,14 +1621,12 @@ describe('runPRWorkflow', () => { await runPRWorkflow(mockOctokit, createDefaultInputs(), 'pull_request', EVENT_PAYLOAD_PATH, FIXTURES_DIR); expect(mockRunSkillTask).toHaveBeenCalledTimes(1); - const [taskOptions, fileConcurrency, _callbacks, semaphore] = mockRunSkillTask.mock.calls[0]!; + const [taskOptions, _callbacks, analysisQueue] = mockRunSkillTask.mock.calls[0]!; expect(taskOptions).toEqual(expect.objectContaining({ name: 'test-skill', displayName: 'test-skill', })); - // When a semaphore is provided, fileConcurrency is unlimited (semaphore is the gate) - expect(fileConcurrency).toBe(Number.MAX_SAFE_INTEGER); - expect(semaphore).toBeInstanceOf(Semaphore); + expect(analysisQueue).toBeInstanceOf(AsyncWorkQueue); }); it('writes a live snapshot after the trigger completes, carrying skillExecutionId and skippedTriggers', async () => { @@ -1693,6 +1691,7 @@ describe('runPRWorkflow', () => { let activeRuns = 0; let maxActiveRuns = 0; let invocationCount = 0; + const analysisQueues: (AsyncWorkQueue | undefined)[] = []; let resolveFirstRun!: () => void; let resolveFirstRunStarted!: () => void; const firstRun = new Promise((resolve) => { @@ -1702,7 +1701,8 @@ describe('runPRWorkflow', () => { resolveFirstRunStarted = resolve; }); - mockRunSkillTask.mockImplementation(async (taskOptions) => { + mockRunSkillTask.mockImplementation(async (taskOptions, _callbacks, analysisQueue) => { + analysisQueues.push(analysisQueue); invocationCount++; activeRuns++; maxActiveRuns = Math.max(maxActiveRuns, activeRuns); @@ -1742,6 +1742,10 @@ describe('runPRWorkflow', () => { expect(mockRunSkillTask).toHaveBeenCalledTimes(2); expect(callsBeforeFirstRunFinished).toBe(1); expect(maxActiveRuns).toBe(1); + expect(analysisQueues).toHaveLength(2); + expect(analysisQueues[0]).toBeInstanceOf(AsyncWorkQueue); + expect(analysisQueues[1]).toBe(analysisQueues[0]); + expect(analysisQueues[0]?.concurrency).toBe(1); }); it('accounts for a trigger the circuit breaker aborted before dispatch, and fails the run since nothing succeeded', async () => { @@ -2225,7 +2229,7 @@ describe('runPRWorkflow', () => { await runPRWorkflow(mockOctokit, createDefaultInputs(), 'pull_request', EVENT_PAYLOAD_PATH, FIXTURES_DIR); // runSkillTask receives options with context containing the custom files - const [taskOptions, fileConcurrency, _callbacks, semaphore] = mockRunSkillTask.mock.calls[0]!; + const [taskOptions] = mockRunSkillTask.mock.calls[0]!; expect(taskOptions.context.pullRequest?.files).toEqual( expect.arrayContaining([ expect.objectContaining({ @@ -2234,8 +2238,6 @@ describe('runPRWorkflow', () => { }), ]) ); - expect(fileConcurrency).toBe(Number.MAX_SAFE_INTEGER); - expect(semaphore).toBeInstanceOf(Semaphore); }); }); diff --git a/packages/warden/src/action/workflow/pr-workflow.ts b/packages/warden/src/action/workflow/pr-workflow.ts index 66c08bc9a..90eabc94c 100644 --- a/packages/warden/src/action/workflow/pr-workflow.ts +++ b/packages/warden/src/action/workflow/pr-workflow.ts @@ -31,7 +31,7 @@ import type { ExistingComment } from '../../output/dedup.js'; import { buildAnalyzedScope, findStaleComments, resolveStaleComments } from '../../output/stale.js'; import { filterFindings } from '../../types/index.js'; import type { EventContext, SkillReport, Finding } from '../../types/index.js'; -import { runPool, Semaphore } from '../../utils/index.js'; +import { AsyncWorkQueue, runPool } from '../../utils/index.js'; import { evaluateFixAttempts, postThreadReply } from '../fix-evaluation/index.js'; import type { EvaluateFixAttemptsResult, FixEvaluation } from '../fix-evaluation/index.js'; import { aggregateUsage } from '../../sdk/usage.js'; @@ -618,12 +618,12 @@ async function executeAllTriggers( const concurrency = runnerConcurrency ?? inputs.parallel; const runtimeEnv = await prepareRuntimeEnvironment(matchedTriggers, inputs); - const semaphore = new Semaphore(concurrency); + const analysisQueue = new AsyncWorkQueue(concurrency); const abortController = new AbortController(); const circuitBreaker = new ProviderFailureCircuitBreaker({ abortController }); const completedSoFar: TriggerResult[] = []; - // Limit trigger dispatch too; the semaphore only gates work after a trigger starts. + // Limit trigger dispatch too; the analysis queue only gates work after a trigger starts. const results = await runPool( matchedTriggers, concurrency, @@ -637,7 +637,7 @@ async function executeAllTriggers( globalMaxFindings: inputs.maxFindings, globalRequestChanges: inputs.requestChanges, globalFailCheck: inputs.failCheck, - semaphore, + analysisQueue, abortController, circuitBreaker, checks: options.checks, diff --git a/packages/warden/src/action/workflow/schedule.test.ts b/packages/warden/src/action/workflow/schedule.test.ts index a4167eab2..3318320a1 100644 --- a/packages/warden/src/action/workflow/schedule.test.ts +++ b/packages/warden/src/action/workflow/schedule.test.ts @@ -338,7 +338,7 @@ describe('runScheduleWorkflow', () => { ]); }); - it('passes auxiliaryMaxRetries through resolved schedule triggers', async () => { + it('passes runner and auxiliary settings through resolved schedule triggers', async () => { mockRunSkill.mockResolvedValue(createSkillReport()); mockBuildContext.mockResolvedValue(createScheduleContext()); @@ -354,12 +354,12 @@ describe('runScheduleWorkflow', () => { expect(mockRunSkill).toHaveBeenNthCalledWith(1, expect.anything(), expect.anything(), - expect.objectContaining({ auxiliaryMaxRetries: 7 }) + expect.objectContaining({ auxiliaryMaxRetries: 7, concurrency: 2 }) ); expect(mockRunSkill).toHaveBeenNthCalledWith(2, expect.anything(), expect.anything(), - expect.objectContaining({ auxiliaryMaxRetries: 3 }) + expect.objectContaining({ auxiliaryMaxRetries: 3, concurrency: 2 }) ); }); diff --git a/packages/warden/src/action/workflow/schedule.ts b/packages/warden/src/action/workflow/schedule.ts index 66020fb3a..1076e28c8 100644 --- a/packages/warden/src/action/workflow/schedule.ts +++ b/packages/warden/src/action/workflow/schedule.ts @@ -120,6 +120,7 @@ async function runScheduleWorkflowInner( logGroupEnd(); let scheduleTriggers: ResolvedTrigger[]; + let runnerConcurrency: number | undefined; let skillRootsByName: LayeredSkillRootsByName | undefined; let service = resolveActionServiceOptions(inputs); try { @@ -133,6 +134,10 @@ async function runScheduleWorkflowInner( || layered.repoConfig?.defaults?.offline === true || layered.config.defaults?.offline === true, ); + runnerConcurrency = + layered.baseConfig?.runner?.concurrency ?? + layered.repoConfig?.runner?.concurrency ?? + layered.config.runner?.concurrency; skillRootsByName = buildSkillRootsByName(repoPath, layered, inputs.baseSkillRoot); service = resolveActionServiceOptions(inputs, layered.config.service); scheduleTriggers = resolveLayeredSkillConfigs(layered, undefined, skillRootsByName) @@ -301,6 +306,7 @@ async function runScheduleWorkflowInner( auxiliaryEffort: resolved.auxiliaryEffort, synthesisModel: resolved.synthesisModel, maxTurns: resolved.maxTurns, + concurrency: runnerConcurrency ?? inputs.parallel, batchDelayMs: resolved.batchDelayMs, maxContextFiles: resolved.maxContextFiles, ignore: resolved.ignore, diff --git a/packages/warden/src/cli/args.ts b/packages/warden/src/cli/args.ts index 990b81857..4e40e0c3c 100644 --- a/packages/warden/src/cli/args.ts +++ b/packages/warden/src/cli/args.ts @@ -27,7 +27,7 @@ export const CLIOptionsSchema = z.object({ /** Only show findings at or above this confidence in output */ minConfidence: ConfidenceThresholdSchema.optional(), help: z.boolean().default(false), - /** Max concurrent file analyses across running skills (default depends on command) */ + /** Max concurrent hunk analyses across running skills (default depends on command) */ parallel: z.number().int().positive().optional(), /** Model to use for analysis (fallback when not set in config) */ model: z.string().optional(), diff --git a/packages/warden/src/cli/help.ts b/packages/warden/src/cli/help.ts index d24c23870..36e81a9da 100644 --- a/packages/warden/src/cli/help.ts +++ b/packages/warden/src/cli/help.ts @@ -164,7 +164,7 @@ const HELP_OPTIONS: Record = { }, parallel: { label: '--parallel ', - description: 'Max concurrent file analyses across running skills', + description: 'Max concurrent hunk analyses across running skills', }, failFast: { label: '-x, --fail-fast', diff --git a/packages/warden/src/cli/main.ts b/packages/warden/src/cli/main.ts index 4736b8809..f3f734fc8 100644 --- a/packages/warden/src/cli/main.ts +++ b/packages/warden/src/cli/main.ts @@ -1424,7 +1424,7 @@ export async function runSkills( return 1; } let tasks: SkillTaskOptions[]; - const concurrency = options.parallel ?? DEFAULT_CONCURRENCY; + const concurrency = options.parallel ?? config?.runner?.concurrency ?? DEFAULT_CONCURRENCY; try { tasks = await createSkillTasks({ specs, diff --git a/packages/warden/src/cli/output/ink-runner.tsx b/packages/warden/src/cli/output/ink-runner.tsx index eacae891b..bcb5ad423 100644 --- a/packages/warden/src/cli/output/ink-runner.tsx +++ b/packages/warden/src/cli/output/ink-runner.tsx @@ -26,7 +26,7 @@ import { type FileState, } from './tasks.js'; import { formatDuration, formatCost, truncate, countBySeverity, formatSeverityDot, pluralize, totalAuxiliaryCost } from './formatters.js'; -import { Semaphore } from '../../utils/index.js'; +import { AsyncWorkQueue } from '../../utils/index.js'; import { Verbosity } from './verbosity.js'; import { ICON_CHECK, ICON_SKIPPED, ICON_PENDING, ICON_ERROR, SPINNER_FRAMES } from './icons.js'; import figures from 'figures'; @@ -276,8 +276,8 @@ export async function runSkillTasksWithInk( : undefined; if (tasks.length === 0 || verbosity === Verbosity.Quiet) { - // No tasks or quiet mode - run without UI using global semaphore. - const semaphore = new Semaphore(concurrency); + // No tasks or quiet mode - run without UI using the global analysis queue. + const analysisQueue = new AsyncWorkQueue(concurrency); const circuitAbortController = new AbortController(); const circuitBreaker = new ProviderFailureCircuitBreaker({ abortController: circuitAbortController }); const composedTasks = composeTasksWithFailFast( @@ -308,7 +308,7 @@ export async function runSkillTasksWithInk( } : {}), }; - return runComposedSkillTasks(composedTasks, callbacks, semaphore); + return runComposedSkillTasks(composedTasks, callbacks, analysisQueue); } // Track skill states @@ -474,8 +474,7 @@ export async function runSkillTasksWithInk( : undefined, }; - // Global semaphore gates file-level work across all skills. - const semaphore = new Semaphore(concurrency); + const analysisQueue = new AsyncWorkQueue(concurrency); // Compose per-task abort controllers: fire on SIGINT, fail-fast, or provider circuit breaker. const circuitAbortController = new AbortController(); @@ -487,8 +486,8 @@ export async function runSkillTasksWithInk( circuitAbortController, ); - // Launch all skills in parallel; the semaphore is the sole concurrency gate. - const results = await runComposedSkillTasks(composedTasks, callbacks, semaphore); + // Launch all skills in parallel; the queue is the sole concurrency gate. + const results = await runComposedSkillTasks(composedTasks, callbacks, analysisQueue); // Flush any pending setImmediate from updateUI so last-tick warnings are // rendered before we tear down. setImmediate is FIFO, so our callback runs diff --git a/packages/warden/src/cli/output/tasks.test.ts b/packages/warden/src/cli/output/tasks.test.ts index 656536a22..065487015 100644 --- a/packages/warden/src/cli/output/tasks.test.ts +++ b/packages/warden/src/cli/output/tasks.test.ts @@ -8,7 +8,6 @@ import type { SkillTaskOptions } from './tasks.js'; import type { FileAnalysisResult } from '../../sdk/types.js'; import type { HunkWithContext } from '../../diff/index.js'; import type { SkillDefinition } from '../../config/schema.js'; -import { Semaphore, runPool } from '../../utils/index.js'; import { SkillRunnerError, WardenAuthenticationError, type ProviderErrorContext } from '../../sdk/errors.js'; import { ProviderFailureCircuitBreaker } from '../../sdk/circuit-breaker.js'; import * as sdkRunner from '../../sdk/runner.js'; @@ -660,6 +659,75 @@ describe('runSkillTasks', () => { expect(resolveSkill).not.toHaveBeenCalled(); }); + it('shares the hunk concurrency limit across skills', async () => { + const fakeHunk = { hunk: { newStart: 1, newCount: 1 } } as unknown as HunkWithContext; + const releases: (() => void)[] = []; + let active = 0; + let maxActive = 0; + + const prepareFiles = vi.spyOn(sdkRunner, 'prepareFiles').mockReturnValue({ + files: [{ filename: 'src/example.ts', hunks: [fakeHunk] }], + skippedFiles: [], + }); + const analyzeFile = vi.spyOn(sdkRunner, 'analyzeFile').mockImplementation( + async (_skill, prepared, _repoPath, _options, _callbacks, _prContext, analysisQueue) => { + expect(analysisQueue).toBeDefined(); + await analysisQueue!.run(async () => { + active++; + maxActive = Math.max(maxActive, active); + await new Promise((resolve) => { + releases.push(() => { + active--; + resolve(); + }); + }); + }); + return { + filename: prepared.filename, + findings: [], + usage: { inputTokens: 1, outputTokens: 1, costUSD: 0 }, + failedHunks: 0, + failedExtractions: 0, + hunkFailures: [], + }; + }, + ); + const postProcessFindings = vi.spyOn(sdkRunner, 'postProcessFindings').mockResolvedValue({ + findings: [], + auxiliaryUsage: [], + }); + const context = { + eventType: 'pull_request', + repository: { owner: 'o', name: 'n', fullName: 'o/n', defaultBranch: 'main' }, + repoPath: '/tmp', + pullRequest: { number: 1, title: 't', body: '', headSha: 'abc', baseSha: 'def', files: [] }, + } as unknown as SkillTaskOptions['context']; + const tasks = ['skill-a', 'skill-b'].map((name) => ({ + name, + resolveSkill: async () => ({ name, description: 'Review.', prompt: 'Review.' }), + context, + })); + + const run = runSkillTasks(tasks, { + mode: logMode(), + verbosity: Verbosity.Quiet, + concurrency: 1, + }); + + await vi.waitFor(() => expect(releases).toHaveLength(1)); + releases.splice(0).forEach((release) => release()); + await vi.waitFor(() => expect(releases).toHaveLength(1)); + releases.splice(0).forEach((release) => release()); + + const results = await run; + expect(maxActive).toBe(1); + expect(results.map((result) => result.name)).toEqual(['skill-a', 'skill-b']); + + prepareFiles.mockRestore(); + analyzeFile.mockRestore(); + postProcessFindings.mockRestore(); + }); + it('does not fail-fast on findings rejected by post-processing', async () => { const candidate = makeFinding(); const controller = new AbortController(); @@ -906,7 +974,7 @@ describe('runSkillTask all-hunks-fail synthesis', () => { }; const onSkillError = vi.fn(); - const result = await runSkillTask(options, 1, { ...noopCallbacks(), onSkillError }); + const result = await runSkillTask(options, { ...noopCallbacks(), onSkillError }); expect(result.report).toBeDefined(); expect(result.report!.error?.code).toBe('auth_failed'); @@ -967,7 +1035,7 @@ describe('runSkillTask all-hunks-fail synthesis', () => { } as unknown as SkillTaskOptions['context'], }; - const result = await runSkillTask(options, 1, noopCallbacks()); + const result = await runSkillTask(options, noopCallbacks()); expect(result.report!.error?.code).toBe('auth_failed'); expect(result.report!.failedHunks).toBe(1); @@ -1028,7 +1096,7 @@ describe('runSkillTask all-hunks-fail synthesis', () => { }); options.runnerOptions = { circuitBreaker }; - const result = await runSkillTask(options, 1, noopCallbacks()); + const result = await runSkillTask(options, noopCallbacks()); expect(result.report!.error?.code).toBe('provider_unavailable'); expect(result.report!.error?.message).toContain('Provider unavailable'); @@ -1091,7 +1159,7 @@ describe('runSkillTask all-hunks-fail synthesis', () => { pullRequest: { number: 1, title: 't', body: '', headSha: 'abc', baseSha: 'def', files: [] }, } as unknown as SkillTaskOptions['context'], runnerOptions: { circuitBreaker }, - }, 1, noopCallbacks()); + }, noopCallbacks()); expect((result.error as SkillRunnerError).code).toBe('provider_unavailable'); expect((result.error as SkillRunnerError).providerContext).toBeUndefined(); @@ -1136,7 +1204,7 @@ describe('runSkillTask all-hunks-fail synthesis', () => { } as unknown as SkillTaskOptions['context'], }; - const result = await runSkillTask(options, 1, noopCallbacks()); + const result = await runSkillTask(options, noopCallbacks()); expect(result.report!.error?.code).toBe('invalid_model_selector'); expect(result.report!.error?.message).toContain('provider/model format'); @@ -1179,7 +1247,7 @@ describe('runSkillTask all-hunks-fail synthesis', () => { }; const onSkillError = vi.fn(); - const result = await runSkillTask(options, 1, { ...noopCallbacks(), onSkillError }); + const result = await runSkillTask(options, { ...noopCallbacks(), onSkillError }); expect(result.error).toBeUndefined(); expect(result.report).toBeDefined(); @@ -1244,7 +1312,7 @@ describe('runSkillTask all-hunks-fail synthesis', () => { } as unknown as SkillTaskOptions['context'], }; - const result = await runSkillTask(options, 1, noopCallbacks()); + const result = await runSkillTask(options, noopCallbacks()); expect(result.report).toBeDefined(); expect(result.report!.error?.code).toBe('extraction_llm_timeout'); @@ -1294,7 +1362,7 @@ describe('runSkillTask all-hunks-fail synthesis', () => { }; const onSkillError = vi.fn(); - const result = await runSkillTask(options, 1, { ...noopCallbacks(), onSkillError }); + const result = await runSkillTask(options, { ...noopCallbacks(), onSkillError }); expect(result.error).toBeUndefined(); expect(result.report).toBeDefined(); @@ -1339,7 +1407,7 @@ describe('runSkillTask skipped path', () => { const onSkillSkipped = vi.fn(); const onSkillComplete = vi.fn(); - const result = await runSkillTask(options, 1, { + const result = await runSkillTask(options, { ...noopCallbacks(), onSkillSkipped, onSkillComplete, @@ -1411,7 +1479,7 @@ describe('runSkillTask model lanes', () => { auxiliaryEffort: 'high', synthesisModel: 'claude-opus-4-5', }, - }, 1, noopCallbacks()); + }, noopCallbacks()); expect(result.report?.error).toBeUndefined(); expect(postProcessSpy).toHaveBeenCalledWith( @@ -1463,7 +1531,7 @@ describe('runSkillTask model lanes', () => { runnerOptions: { runtime: 'pi', }, - }, 1, noopCallbacks()); + }, noopCallbacks()); expect(result.report?.model).toBe('claude-sonnet-4-5-20260929'); }); @@ -1502,7 +1570,7 @@ describe('runSkillTask model lanes', () => { runnerOptions: { runtime: 'pi', }, - }, 1, noopCallbacks()); + }, noopCallbacks()); expect(result.report?.error).toBeDefined(); expect(result.report?.model).toBe('claude-sonnet-4-5-20260929'); @@ -1539,7 +1607,7 @@ describe('runSkillTask error capture', () => { }; const onSkillError = vi.fn(); - const result = await runSkillTask(options, 1, { ...noopCallbacks(), onSkillError }); + const result = await runSkillTask(options, { ...noopCallbacks(), onSkillError }); expect(result.name).toBe('auth-skill'); expect(result.report).toBeDefined(); @@ -1561,7 +1629,7 @@ describe('runSkillTask error capture', () => { context: makeContext(), }; - const result = await runSkillTask(options, 1, noopCallbacks()); + const result = await runSkillTask(options, noopCallbacks()); expect(result.report).toBeDefined(); expect(result.report!.error?.code).toBe('skill_resolution_failed'); @@ -1580,41 +1648,9 @@ describe('runSkillTask error capture', () => { context: makeContext(), }; - const result = await runSkillTask(options, 1, noopCallbacks()); + const result = await runSkillTask(options, noopCallbacks()); expect(result.failOn).toBe('high'); expect(result.minConfidence).toBe('medium'); }); }); - -describe('Semaphore integration with runPool', () => { - it('limits concurrent file analyses across skills to the semaphore size', async () => { - // Track concurrent active file analyses - let active = 0; - let maxActive = 0; - const concurrencyLimit = 2; - const semaphore = new Semaphore(concurrencyLimit); - - // Simulate 3 skills each with 3 files (9 total file analyses). - // runPool gets unlimited concurrency (like skills launching in parallel), - // but the semaphore gates how many run simultaneously. - const fileWork = Array.from({ length: 9 }, (_, i) => i); - - const results = await runPool(fileWork, fileWork.length, async (item) => { - await semaphore.acquire(); - try { - active++; - maxActive = Math.max(maxActive, active); - await new Promise((resolve) => setTimeout(resolve, 5)); - active--; - return item; - } finally { - semaphore.release(); - } - }); - - expect(results).toHaveLength(9); - expect(maxActive).toBeLessThanOrEqual(concurrencyLimit); - expect(maxActive).toBe(concurrencyLimit); - }); -}); diff --git a/packages/warden/src/cli/output/tasks.ts b/packages/warden/src/cli/output/tasks.ts index 20c0f3aac..d7009f0f2 100644 --- a/packages/warden/src/cli/output/tasks.ts +++ b/packages/warden/src/cli/output/tasks.ts @@ -27,6 +27,7 @@ import { type FindingProcessingEvent, } from '../../sdk/runner.js'; import { ProviderFailureCircuitBreaker } from '../../sdk/circuit-breaker.js'; +import { DEFAULT_ANALYSIS_CONCURRENCY } from '../../sdk/types.js'; import { buildFileReports } from '../../sdk/report-files.js'; import chalk from 'chalk'; import figures from 'figures'; @@ -35,7 +36,7 @@ import type { OutputMode } from './tty.js'; import { ICON_CHECK, ICON_SKIPPED } from './icons.js'; import { timestamp } from './tty.js'; import { formatDuration, formatCost, formatLocation, formatSeverityPlain, formatFindingCountsPlain, countBySeverity, pluralize } from './formatters.js'; -import { runPool, Semaphore } from '../../utils/index.js'; +import { AsyncWorkQueue, runPool } from '../../utils/index.js'; /** * Result from processing a single file within a skill task. @@ -273,9 +274,8 @@ export interface SkillProgressCallbacks { */ export async function runSkillTask( options: SkillTaskOptions, - fileConcurrency: number, callbacks: SkillProgressCallbacks, - semaphore?: Semaphore + sharedAnalysisQueue?: AsyncWorkQueue, ): Promise { const { name, @@ -289,6 +289,10 @@ export async function runSkillTask( } = options; // This clone's identity scopes circuit-breaker provider diagnostics to this skill run. const runnerOptions: SkillRunnerOptions = { ...configuredRunnerOptions }; + const taskConcurrency = runnerOptions.parallel === false + ? 1 + : runnerOptions.concurrency ?? DEFAULT_ANALYSIS_CONCURRENCY; + const analysisQueue = sharedAnalysisQueue ?? new AsyncWorkQueue(taskConcurrency); return Sentry.startSpan( { op: 'skill.run', name: `run ${displayName}` }, @@ -387,29 +391,30 @@ export async function runSkillTask( } : undefined; - // Process files with concurrency + // Files aggregate hunk results; the shared queue owns concurrency. const processFile = async (prepared: PreparedFile, index: number): Promise => { const filename = prepared.filename; - const fileStartTime = Date.now(); - - // Update file state to running (local + callback) const localState = fileStates[index]; - if (localState) localState.status = 'running'; - callbacks.onFileUpdate(name, filename, { status: 'running' }); + let fileStartTime: number | undefined; const fileCallbacks: FileAnalysisCallbacks = { skillStartTime: startTime, onHunkStart: (hunkNum, totalHunks, lineRange) => { - callbacks.onFileUpdate(name, filename, { - currentHunk: hunkNum, - totalHunks, - }); + if (fileStartTime === undefined) { + fileStartTime = Date.now(); + if (localState) localState.status = 'running'; + callbacks.onFileUpdate(name, filename, { + status: 'running', + totalHunks, + }); + } callbacks.onHunkStart?.(name, filename, hunkNum, totalHunks, lineRange); }, onHunkComplete: (_hunkNum, findings, usage) => { // Accumulate findings and usage for this file const current = fileStates[index]; if (current) { + current.currentHunk++; current.findings.push(...findings); if (current.usage) { current.usage.inputTokens += usage.inputTokens; @@ -433,7 +438,10 @@ export async function runSkillTask( } else { current.usage = { ...usage }; } - callbacks.onFileUpdate(name, filename, { usage: current.usage }); + callbacks.onFileUpdate(name, filename, { + currentHunk: current.currentHunk, + usage: current.usage, + }); } }, onLargePrompt: callbacks.onLargePrompt @@ -479,11 +487,12 @@ export async function runSkillTask( context.repoPath, runnerOptions, fileCallbacks, - prContext + prContext, + analysisQueue, ); // Detect if this file was aborted before any real work happened - const fileDurationMs = Date.now() - fileStartTime; + const fileDurationMs = fileStartTime === undefined ? 0 : Date.now() - fileStartTime; const aborted = runnerOptions.abortController?.signal.aborted ?? false; const noWork = !result.usage || (result.usage.inputTokens === 0 && result.usage.outputTokens === 0); const fileStatus = (aborted && noWork) ? 'skipped' : 'done'; @@ -509,38 +518,9 @@ export async function runSkillTask( }; }; - // Return an empty result for files skipped due to abort - const processSkippedFile = (index: number): FileProcessResult => { - const localState = fileStates[index]; - if (localState) localState.status = 'skipped'; - const filename = preparedFiles[index]?.filename ?? 'unknown'; - callbacks.onFileUpdate(name, filename, { status: 'skipped' }); - return { findings: [], durationMs: 0, failedHunks: 0, failedExtractions: 0, hunkFailures: [] }; - }; - - // Process files with sliding-window concurrency pool - const batchDelayMs = runnerOptions.batchDelayMs ?? 0; - const shouldAbort = () => runnerOptions.abortController?.signal.aborted ?? false; - // The effective concurrency for batch delay: when a semaphore gates work, - // use its permit count (the actual concurrency limit) rather than fileConcurrency. - const effectiveConcurrency = semaphore ? semaphore.initialPermits : fileConcurrency; - const allResults = await runPool(preparedFiles, fileConcurrency, - async (file, index) => { - if (semaphore) await semaphore.acquire(); - try { - // Check abort after acquiring the semaphore -- the file may have - // been queued behind others and a SIGINT could have arrived while waiting. - if (shouldAbort()) return processSkippedFile(index); - // Rate-limit: delay items beyond the first concurrent wave - if (index >= effectiveConcurrency && batchDelayMs > 0) { - await new Promise((resolve) => setTimeout(resolve, batchDelayMs)); - } - return await processFile(file, index); - } finally { - if (semaphore) semaphore.release(); - } - }, - { shouldAbort } + // Files only group results and progress. The shared queue schedules every hunk. + const allResults = await Promise.all( + preparedFiles.map((file, index) => processFile(file, index)), ); // Mark never-dispatched files as skipped @@ -1022,15 +1002,15 @@ export function composeTasksWithFailFast( } /** - * Launch all skill tasks in parallel using a shared semaphore for concurrency. + * Launch all skill tasks in parallel using a shared hunk queue. */ export async function runComposedSkillTasks( tasks: SkillTaskOptions[], callbacks: SkillProgressCallbacks, - semaphore: Semaphore + analysisQueue: AsyncWorkQueue, ): Promise { const results = await runPool(tasks, tasks.length, - (task) => runSkillTask(task, Number.MAX_SAFE_INTEGER, callbacks, semaphore), + (task) => runSkillTask(task, callbacks, analysisQueue), { shouldAbort: () => tasks[0]?.runnerOptions?.abortController?.signal.aborted ?? false } ); @@ -1048,10 +1028,7 @@ export async function runSkillTasks( ): Promise { const { mode, verbosity, concurrency, failFastController, onSkillComplete, onChunkComplete } = options; - // Global semaphore gates file-level work across all skills. - // All skills launch immediately so the UI shows them as "running", - // but only `concurrency` files will be analysed at any time. - const semaphore = new Semaphore(concurrency); + const analysisQueue = new AsyncWorkQueue(concurrency); const effectiveCallbacks = callbacks ?? createDefaultCallbacks(tasks, mode, verbosity); @@ -1110,6 +1087,6 @@ export async function runSkillTasks( }, { once: true }); } - // Launch all skills in parallel; the semaphore is the sole concurrency gate. - return runComposedSkillTasks(composedTasks, wrappedCallbacks, semaphore); + // Launch all skills in parallel; the queue is the sole concurrency gate. + return runComposedSkillTasks(composedTasks, wrappedCallbacks, analysisQueue); } diff --git a/packages/warden/src/config/schema.ts b/packages/warden/src/config/schema.ts index fea4ccfc7..6edd4b600 100644 --- a/packages/warden/src/config/schema.ts +++ b/packages/warden/src/config/schema.ts @@ -162,7 +162,7 @@ export type SkillConfig = z.infer; // Runner configuration export const RunnerConfigSchema = z.object({ - /** Max concurrent file analyses across all skills (default: 4) */ + /** Max concurrent hunk analyses across all skills (default: 4) */ concurrency: z.number().int().positive().optional(), }); export type RunnerConfig = z.infer; @@ -268,7 +268,7 @@ export const DefaultsSchema = z.object({ ignore: IgnoreConfigSchema.optional(), /** Global scan limits applied after ignore filtering */ scan: ScanConfigSchema.optional(), - /** Delay in milliseconds between batch starts when processing files in parallel. Default: 0 */ + /** Delay applied before each analysis dispatched after the first concurrent wave. Default: 0 */ batchDelayMs: z.number().int().nonnegative().optional(), /** Max retries for auxiliary structured model calls (extraction repair, merging, dedup, fix evaluation). Default: 5 */ auxiliaryMaxRetries: z.number().int().positive().optional(), diff --git a/packages/warden/src/sdk/analyze.test.ts b/packages/warden/src/sdk/analyze.test.ts index 0d62803a0..d9ff7c5f3 100644 --- a/packages/warden/src/sdk/analyze.test.ts +++ b/packages/warden/src/sdk/analyze.test.ts @@ -10,6 +10,7 @@ import { ProviderFailureCircuitBreaker } from './circuit-breaker.js'; import { SkillRunnerError } from './errors.js'; import { Sentry } from '../sentry.js'; import { startTracedSpan } from '../sentry-trace.js'; +import { AsyncWorkQueue } from '../utils/index.js'; vi.mock('./runtimes/index.js', () => ({ getRuntime: vi.fn(), @@ -319,6 +320,77 @@ describe('analyzeFile', () => { vi.restoreAllMocks(); }); + it('queues hunks across files while preserving per-file result order', async () => { + const releases: (() => void)[] = []; + const gates = Array.from({ length: 4 }, () => new Promise((resolve) => { + releases.push(resolve); + })); + let nextHunk = 0; + let active = 0; + let maxActive = 0; + const runSkill = vi.fn(async () => { + const hunkIndex = nextHunk++; + active++; + maxActive = Math.max(maxActive, active); + await gates[hunkIndex]; + active--; + const findings = hunkIndex < 3 ? [makeFinding(hunkIndex + 1)] : []; + return { + result: { + status: 'success' as const, + text: JSON.stringify({ findings }), + errors: [], + usage: makeUsage(), + }, + }; + }); + vi.mocked(getRuntime).mockReturnValue({ + name: 'pi', + runSkill, + runAuxiliary: vi.fn(), + runSynthesis: vi.fn(), + } as unknown as Runtime); + const completionOrder: number[] = []; + const queue = new AsyncWorkQueue(2); + const skill = { + name: 'security-review', + description: 'Security review.', + prompt: 'Return findings as JSON.', + }; + + const firstFile = analyzeFile( + skill, + makePreparedFile(3), + '/tmp/repo', + { runtime: 'pi' }, + { onHunkComplete: (hunkNum) => completionOrder.push(hunkNum) }, + undefined, + queue, + ); + const secondFile = analyzeFile( + skill, + makePreparedFile(), + '/tmp/repo', + { runtime: 'pi' }, + undefined, + undefined, + queue, + ); + + await vi.waitFor(() => expect(runSkill).toHaveBeenCalledTimes(2)); + releases[1]!(); + await vi.waitFor(() => expect(runSkill).toHaveBeenCalledTimes(3)); + releases[2]!(); + await vi.waitFor(() => expect(runSkill).toHaveBeenCalledTimes(4)); + releases[0]!(); + releases[3]!(); + + const [result] = await Promise.all([firstFile, secondFile]); + expect(completionOrder).toEqual([2, 3, 1]); + expect(result.findings.map((finding) => finding.location?.startLine)).toEqual([1, 2, 3]); + expect(maxActive).toBe(2); + }); + it('retries extraction once after a transient auxiliary failure', async () => { const runAuxiliary = vi.fn() .mockResolvedValueOnce({ success: false, error: 'malformed tool call' }) @@ -438,6 +510,7 @@ describe('analyzeFile', () => { { abortController: controller, circuitBreaker, + concurrency: 1, retry: { maxRetries: 0, initialDelayMs: 1, @@ -482,6 +555,7 @@ describe('analyzeFile', () => { { abortController: controller, circuitBreaker, + concurrency: 1, retry: { maxRetries: 0, initialDelayMs: 1, @@ -836,6 +910,74 @@ describe('runSkill', () => { expect(report.runtime).toBe('pi'); }); + it('shares hunk concurrency across files and reports the active file lifecycle', async () => { + const releases: (() => void)[] = []; + let active = 0; + let maxActive = 0; + const runSkillMock = vi.fn(async () => { + active++; + maxActive = Math.max(maxActive, active); + await new Promise((resolve) => { + releases.push(() => { + active--; + resolve(); + }); + }); + return { + result: { + status: 'success' as const, + text: JSON.stringify({ findings: [] }), + errors: [], + usage: makeUsage(), + }, + }; + }); + vi.mocked(getRuntime).mockReturnValue({ + name: 'pi', + runSkill: runSkillMock, + runAuxiliary: vi.fn(), + runSynthesis: vi.fn(), + } as unknown as Runtime); + const context = makeContextWithOneHunk(); + const firstFile = context.pullRequest!.files[0]!; + context.pullRequest!.files = [ + firstFile, + { ...firstFile, filename: 'src/other.ts' }, + ]; + const startedFiles: string[] = []; + const completedFiles: string[] = []; + + const report = runSkill( + { + name: 'security-review', + description: 'Security review.', + prompt: 'Return findings as JSON.', + }, + context, + { + runtime: 'pi', + concurrency: 1, + verifyFindings: false, + callbacks: { + onFileStart: (filename) => startedFiles.push(filename), + onFileComplete: (filename) => completedFiles.push(filename), + }, + }, + ); + + await vi.waitFor(() => expect(runSkillMock).toHaveBeenCalledTimes(1)); + expect(startedFiles).toEqual(['src/example.ts']); + expect(completedFiles).toEqual([]); + releases.shift()!(); + await vi.waitFor(() => expect(runSkillMock).toHaveBeenCalledTimes(2)); + releases.shift()!(); + await report; + + expect(maxActive).toBe(1); + expect(startedFiles).toEqual(['src/example.ts', 'src/other.ts']); + expect(completedFiles).toEqual(['src/example.ts', 'src/other.ts']); + }); + it('reports the model that actually answered when no model override is configured', async () => { const runSkillMock = vi.fn().mockResolvedValue({ result: { diff --git a/packages/warden/src/sdk/analyze.ts b/packages/warden/src/sdk/analyze.ts index ee36df7f4..8bb415608 100644 --- a/packages/warden/src/sdk/analyze.ts +++ b/packages/warden/src/sdk/analyze.ts @@ -18,7 +18,7 @@ import { getRuntime, getRuntimeProviderOptions } from './runtimes/index.js'; import type { SkillRunResult } from './runtimes/index.js'; import { LARGE_PROMPT_THRESHOLD_CHARS, - DEFAULT_FILE_CONCURRENCY, + DEFAULT_ANALYSIS_CONCURRENCY, type AuxiliaryUsageEntry, type HunkAnalysisResult, type HunkAnalysisCallbacks, @@ -31,7 +31,7 @@ import { import { prepareFiles } from './prepare.js'; import type { EventContext, SkillReport, UsageStats, HunkFailure, HunkTrace, VerifierRejections } from '../types/index.js'; import type { SourceSnippet, SourceSnippetLine } from '../types/index.js'; -import { runPool } from '../utils/index.js'; +import { AsyncWorkQueue } from '../utils/index.js'; import { getSpanContext, startTraceRecorder, withTraceRecorder, type TraceRecorder } from '../sentry-trace.js'; /** Result from parsing hunk output */ @@ -324,7 +324,8 @@ async function analyzeHunk( repoPath: string, options: SkillRunnerOptions, callbacks?: HunkAnalysisCallbacks, - prContext?: PRPromptContext + prContext?: PRPromptContext, + parentSpan?: Span, ): Promise { if (options.captureTraces) { ensureLocalTracing(); @@ -336,6 +337,7 @@ async function analyzeHunk( { op: 'skill.analyze_hunk', name: `analyze hunk ${hunkCtx.filename}:${lineRange}`, + ...(parentSpan ? { parentSpan } : {}), attributes: { 'gen_ai.agent.name': skill.name, 'code.file.path': hunkCtx.filename, @@ -828,7 +830,8 @@ export async function analyzeFile( repoPath: string, options: SkillRunnerOptions = {}, callbacks?: FileAnalysisCallbacks, - prContext?: PRPromptContext + prContext?: PRPromptContext, + analysisQueue?: AsyncWorkQueue, ): Promise { return Sentry.startSpan( { @@ -851,27 +854,72 @@ export async function analyzeFile( let failedHunks = 0; let failedExtractions = 0; - for (const [hunkIndex, hunk] of file.hunks.entries()) { - if (abortController?.signal.aborted) break; + const concurrency = options.parallel === false + ? 1 + : options.concurrency ?? DEFAULT_ANALYSIS_CONCURRENCY; + const queue = analysisQueue ?? new AsyncWorkQueue(concurrency); + const batchDelayMs = options.parallel === false ? 0 : options.batchDelayMs; + const completedHunks = await Promise.all(file.hunks.map((hunk, hunkIndex) => + queue.run(async () => { + if (abortController?.signal.aborted) return undefined; - const lineRange = formatHunkLineRange(hunk); - callbacks?.onHunkStart?.(hunkIndex + 1, file.hunks.length, lineRange); + const lineRange = formatHunkLineRange(hunk); + callbacks?.onHunkStart?.(hunkIndex + 1, file.hunks.length, lineRange); - const hunkCallbacks: HunkAnalysisCallbacks | undefined = callbacks - ? { - lineRange, - onLargePrompt: callbacks.onLargePrompt, - onPromptSize: callbacks.onPromptSize, - onRetry: callbacks.onRetry, - onExtractionFailure: callbacks.onExtractionFailure, - onExtractionResult: callbacks.onExtractionResult, - onHunkFailed: callbacks.onHunkFailed, - } - : undefined; + const hunkCallbacks: HunkAnalysisCallbacks | undefined = callbacks + ? { + lineRange, + onLargePrompt: callbacks.onLargePrompt, + onPromptSize: callbacks.onPromptSize, + onRetry: callbacks.onRetry, + onExtractionFailure: callbacks.onExtractionFailure, + onExtractionResult: callbacks.onExtractionResult, + onHunkFailed: callbacks.onHunkFailed, + } + : undefined; + + const hunkStartTime = Date.now(); + const result = await analyzeHunk( + skill, + hunk, + repoPath, + options, + hunkCallbacks, + prContext, + span, + ); + const hunkDurationMs = Date.now() - hunkStartTime; + + attachElapsedTime(result.findings, callbacks?.skillStartTime); + callbacks?.onHunkComplete?.(hunkIndex + 1, result.findings, result.usage); + const chunkResult: ChunkAnalysisResult = { + filename: file.filename, + model: options.model, + index: hunkIndex + 1, + total: file.hunks.length, + lineRange, + findings: result.findings, + usage: result.usage, + durationMs: hunkDurationMs, + failed: result.failed && result.failureCode !== 'aborted', + extractionFailed: result.extractionFailed, + failureCode: result.failureCode, + failureMessage: result.failureMessage, + extractionError: result.extractionError, + extractionPreview: result.extractionPreview, + auxiliaryUsage: result.auxiliaryUsage, + trace: result.trace, + }; + callbacks?.onChunkComplete?.(chunkResult); - const hunkStartTime = Date.now(); - const result = await analyzeHunk(skill, hunk, repoPath, options, hunkCallbacks, prContext); - const hunkDurationMs = Date.now() - hunkStartTime; + return { lineRange, result }; + }, { delayMs: batchDelayMs, signal: abortController?.signal }), + )); + + // Promise.all preserves hunk order even when analyses finish out of order. + for (const completed of completedHunks) { + if (!completed) continue; + const { lineRange, result } = completed; // `failed` and `extractionFailed` are conceptually mutually exclusive: // if analysis failed (no output produced), there's nothing to extract. @@ -900,33 +948,12 @@ export async function analyzeFile( }); } - attachElapsedTime(result.findings, callbacks?.skillStartTime); - callbacks?.onHunkComplete?.(hunkIndex + 1, result.findings, result.usage); if (result.trace) { hunkTraces.push(result.trace); } if (result.responseModel) { fileResponseModels.push(result.responseModel); } - const chunkResult: ChunkAnalysisResult = { - filename: file.filename, - model: options.model, - index: hunkIndex + 1, - total: file.hunks.length, - lineRange, - findings: result.findings, - usage: result.usage, - durationMs: hunkDurationMs, - failed: result.failed && result.failureCode !== 'aborted', - extractionFailed: result.extractionFailed, - failureCode: result.failureCode, - failureMessage: result.failureMessage, - extractionError: result.extractionError, - extractionPreview: result.extractionPreview, - auxiliaryUsage: result.auxiliaryUsage, - trace: result.trace, - }; - callbacks?.onChunkComplete?.(chunkResult); fileFindings.push(...result.findings); fileUsage.push(result.usage); @@ -1072,21 +1099,21 @@ async function runSkillAnalysis( maxContextFiles: options.maxContextFiles, }; - /** - * Process all hunks for a single file sequentially. - * Wraps analyzeFile with progress callbacks. - */ + /** Wrap analyzeFile with progress callbacks. */ async function processFile( fileHunkEntry: PreparedFile, fileIndex: number - ): Promise { + ): Promise<{ filename: string; result: FileAnalysisResult; durationMs: number }> { const { filename } = fileHunkEntry; - - callbacks?.onFileStart?.(filename, fileIndex, totalFiles); + let fileStartTime: number | undefined; const fileCallbacks: FileAnalysisCallbacks = { skillStartTime: callbacks?.skillStartTime, onHunkStart: (hunkNum, totalHunks, lineRange) => { + if (fileStartTime === undefined) { + fileStartTime = Date.now(); + callbacks?.onFileStart?.(filename, fileIndex, totalFiles); + } callbacks?.onHunkStart?.(filename, hunkNum, totalHunks, lineRange); }, onHunkComplete: (hunkNum, findings, usage) => { @@ -1124,39 +1151,39 @@ async function runSkillAnalysis( : undefined, }; - const result = await analyzeFile(skill, fileHunkEntry, context.repoPath, options, fileCallbacks, prContext); + const result = await analyzeFile( + skill, + fileHunkEntry, + context.repoPath, + options, + fileCallbacks, + prContext, + analysisQueue, + ); - callbacks?.onFileComplete?.(filename, fileIndex, totalFiles); + if (fileStartTime !== undefined) { + callbacks?.onFileComplete?.(filename, fileIndex, totalFiles); + } - return result; + return { + filename, + result, + durationMs: fileStartTime === undefined ? 0 : Date.now() - fileStartTime, + }; } - /** Process a file with timing, returning a self-contained result. */ - async function processFileWithTiming(fileHunkEntry: PreparedFile, fileIndex: number) { - const fileStart = Date.now(); - const result = await processFile(fileHunkEntry, fileIndex); - const durationMs = Date.now() - fileStart; - return { filename: fileHunkEntry.filename, result, durationMs }; - } + const concurrency = parallel + ? options.concurrency ?? DEFAULT_ANALYSIS_CONCURRENCY + : 1; + const analysisQueue = new AsyncWorkQueue(concurrency); // Collect results in input order (Promise.all preserves order) const fileResults: { filename: string; result: FileAnalysisResult; durationMs: number }[] = []; // Process files - parallel or sequential based on options if (parallel) { - // Process files with sliding-window concurrency pool - const fileConcurrency = options.concurrency ?? DEFAULT_FILE_CONCURRENCY; - const batchDelayMs = options.batchDelayMs ?? 0; - - fileResults.push(...await runPool(fileHunks, fileConcurrency, - async (fileHunkEntry, index) => { - // Rate-limit: delay items beyond the first concurrent wave - if (index >= fileConcurrency && batchDelayMs > 0) { - await new Promise((resolve) => setTimeout(resolve, batchDelayMs)); - } - return processFileWithTiming(fileHunkEntry, index); - }, - { shouldAbort: () => abortController?.signal.aborted ?? false } + fileResults.push(...await Promise.all( + fileHunks.map((fileHunkEntry, index) => processFile(fileHunkEntry, index)), )); } else { // Process files sequentially @@ -1164,7 +1191,7 @@ async function runSkillAnalysis( // Check for abort before starting new file if (abortController?.signal.aborted) break; - fileResults.push(await processFileWithTiming(fileHunkEntry, fileIndex)); + fileResults.push(await processFile(fileHunkEntry, fileIndex)); } } diff --git a/packages/warden/src/sdk/types.ts b/packages/warden/src/sdk/types.ts index 9a06fc012..5425ec9b9 100644 --- a/packages/warden/src/sdk/types.ts +++ b/packages/warden/src/sdk/types.ts @@ -20,8 +20,8 @@ export interface FindingProcessingEvent { replacement?: Finding; } -/** Default concurrency for file-level parallel processing (standalone SDK usage only) */ -export const DEFAULT_FILE_CONCURRENCY = 5; +/** Default concurrency for hunk analysis (standalone SDK usage only). */ +export const DEFAULT_ANALYSIS_CONCURRENCY = 5; /** Threshold in characters above which to warn about large prompts (~25k tokens) */ export const LARGE_PROMPT_THRESHOLD_CHARS = 100000; @@ -104,11 +104,11 @@ export interface SkillRunnerOptions { maxTurns?: number; /** Lines of context to include around each hunk */ contextLines?: number; - /** Process files in parallel (default: true) */ + /** Process hunks in parallel (default: true) */ parallel?: boolean; - /** Max concurrent file analyses when parallel=true (default: 5) */ + /** Max concurrent hunk analyses when parallel=true (default: 5) */ concurrency?: number; - /** Delay in milliseconds between batch starts when parallel=true (default: 0) */ + /** Delay before queued analyses start after the first concurrent wave (default: 0) */ batchDelayMs?: number; /** Model to use for analysis (e.g., 'openai/gpt-5.5'). Uses SDK default if not specified. */ model?: string; diff --git a/packages/warden/src/utils/async.test.ts b/packages/warden/src/utils/async.test.ts index f535fe2db..4f824a79b 100644 --- a/packages/warden/src/utils/async.test.ts +++ b/packages/warden/src/utils/async.test.ts @@ -1,5 +1,9 @@ -import { describe, it, expect, vi } from 'vitest'; -import { runPool, processInBatches, Semaphore } from './async.js'; +import { afterEach, describe, it, expect, vi } from 'vitest'; +import { AsyncWorkQueue, runPool, processInBatches } from './async.js'; + +afterEach(() => { + vi.useRealTimers(); +}); describe('runPool', () => { it('processes all items and returns results in input order', async () => { @@ -136,82 +140,87 @@ describe('processInBatches', () => { }); }); -describe('Semaphore', () => { - it('allows immediate acquisition when permits are available', async () => { - const sem = new Semaphore(2); - await sem.acquire(); - await sem.acquire(); - // Both acquired without blocking - sem.release(); - sem.release(); - }); - - it('blocks when no permits are available and unblocks on release', async () => { - const sem = new Semaphore(1); - await sem.acquire(); - - let acquired = false; - const pending = sem.acquire().then(() => { acquired = true; }); +describe('AsyncWorkQueue', () => { + it('runs dynamically submitted work up to the concurrency limit', async () => { + const queue = new AsyncWorkQueue(3); + let active = 0; + let maxActive = 0; - // Give the microtask queue a tick - await new Promise((resolve) => setTimeout(resolve, 0)); - expect(acquired).toBe(false); + const work = async (value: number) => { + active++; + maxActive = Math.max(maxActive, active); + await new Promise((resolve) => setTimeout(resolve, 10)); + active--; + return value; + }; - sem.release(); - await pending; - expect(acquired).toBe(true); + const results = await Promise.all( + Array.from({ length: 10 }, (_, index) => queue.run(() => work(index))), + ); - sem.release(); + expect(maxActive).toBe(3); + expect(results).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9]); }); - it('wakes waiters in FIFO order', async () => { - const sem = new Semaphore(1); - await sem.acquire(); - - const order: number[] = []; + it('starts queued work in FIFO order', async () => { + const queue = new AsyncWorkQueue(1); + const starts: number[] = []; - const p1 = sem.acquire().then(() => { order.push(1); }); - const p2 = sem.acquire().then(() => { order.push(2); }); - const p3 = sem.acquire().then(() => { order.push(3); }); + await Promise.all([1, 2, 3].map((value) => queue.run(async () => { + starts.push(value); + }))); - sem.release(); // wakes waiter 1 - await p1; - sem.release(); // wakes waiter 2 - await p2; - sem.release(); // wakes waiter 3 - await p3; + expect(starts).toEqual([1, 2, 3]); + }); - expect(order).toEqual([1, 2, 3]); + it('continues dispatching after work rejects', async () => { + const queue = new AsyncWorkQueue(1); + const failed = queue.run(async () => { + throw new Error('failed'); + }); + const completed = queue.run(async () => 'completed'); - sem.release(); + await expect(failed).rejects.toThrow('failed'); + await expect(completed).resolves.toBe('completed'); }); - it('limits concurrent work to the permit count', async () => { - const sem = new Semaphore(3); - let active = 0; - let maxActive = 0; - - const work = async () => { - await sem.acquire(); - active++; - maxActive = Math.max(maxActive, active); - await new Promise((resolve) => setTimeout(resolve, 10)); - active--; - sem.release(); - }; + it('delays work after the first concurrent wave', async () => { + vi.useFakeTimers(); + const queue = new AsyncWorkQueue(1); + const starts: number[] = []; - await Promise.all(Array.from({ length: 10 }, () => work())); + await queue.run(async () => { + starts.push(1); + }); + const delayed = queue.run(async () => { + starts.push(2); + }, { delayMs: 100 }); + + await vi.advanceTimersByTimeAsync(99); + expect(starts).toEqual([1]); + await vi.advanceTimersByTimeAsync(1); + await delayed; + expect(starts).toEqual([1, 2]); + }); + + it('stops waiting for a delayed start when aborted', async () => { + vi.useFakeTimers(); + const controller = new AbortController(); + const queue = new AsyncWorkQueue(1); + + await queue.run(async () => undefined); + const work = vi.fn(async () => undefined); + const delayed = queue.run(work, { + delayMs: 10_000, + signal: controller.signal, + }); - expect(maxActive).toBe(3); + controller.abort(); + await delayed; + expect(work).toHaveBeenCalledOnce(); }); - it('handles release without waiters (restores permits)', async () => { - const sem = new Semaphore(1); - await sem.acquire(); - sem.release(); - - // Should be able to acquire again immediately - await sem.acquire(); - sem.release(); + it('rejects invalid concurrency', () => { + expect(() => new AsyncWorkQueue(0)).toThrow('positive integer'); }); }); diff --git a/packages/warden/src/utils/async.ts b/packages/warden/src/utils/async.ts index 112e811f1..3b2d595d2 100644 --- a/packages/warden/src/utils/async.ts +++ b/packages/warden/src/utils/async.ts @@ -1,39 +1,98 @@ +interface QueuedWork { + work: () => Promise; + delayMs: number; + signal?: AbortSignal; + resolve: (value: T) => void; + reject: (reason: unknown) => void; +} + +export interface QueueWorkOptions { + /** Delay before starting work after the queue's first concurrent wave. */ + delayMs?: number; + /** Stop waiting for the start delay when cancellation is requested. */ + signal?: AbortSignal; +} + /** - * A counting semaphore for limiting concurrent access to a shared resource. - * Callers acquire a permit before starting work and release it when done. - * If no permits are available, acquire() blocks until one is released. + * A FIFO queue for dynamically submitted asynchronous work. + * + * The queue owns concurrency bookkeeping so callers cannot leak permits or + * accidentally bypass the shared limit. Work starts as capacity becomes + * available, and each returned promise settles with its submitted task. */ -export class Semaphore { - private permits: number; - private waiters: (() => void)[] = []; - /** The initial permit count this semaphore was created with. */ - readonly initialPermits: number; - - constructor(permits: number) { - this.permits = permits; - this.initialPermits = permits; - } +export class AsyncWorkQueue { + private readonly pending: QueuedWork[] = []; + private active = 0; + private started = 0; + readonly concurrency: number; - async acquire(): Promise { - if (this.permits > 0) { - this.permits--; - return; + constructor(concurrency: number) { + if (!Number.isInteger(concurrency) || concurrency < 1) { + throw new RangeError('Queue concurrency must be a positive integer'); } - return new Promise((resolve) => { - this.waiters.push(resolve); + this.concurrency = concurrency; + } + + /** Enqueue work under the queue's shared concurrency and delayed-dispatch limits. */ + run(work: () => Promise, options: QueueWorkOptions = {}): Promise { + return new Promise((resolve, reject) => { + this.pending.push({ + work, + delayMs: options.delayMs ?? 0, + signal: options.signal, + resolve: resolve as (value: unknown) => void, + reject, + }); + this.dispatch(); }); } - release(): void { - const next = this.waiters.shift(); - if (next) { - next(); - } else { - this.permits++; + private dispatch(): void { + while (this.active < this.concurrency) { + const item = this.pending.shift(); + if (!item) return; + + const delayMs = this.started >= this.concurrency ? item.delayMs : 0; + this.started++; + this.active++; + + void this.execute(item, delayMs).finally(() => { + this.active--; + this.dispatch(); + }); + } + } + + private async execute(item: QueuedWork, delayMs: number): Promise { + try { + if (delayMs > 0) { + await waitForDelay(delayMs, item.signal); + } + item.resolve(await item.work()); + } catch (error) { + item.reject(error); } } } +function waitForDelay(delayMs: number, signal?: AbortSignal): Promise { + return new Promise((resolve) => { + if (signal?.aborted) { + resolve(); + return; + } + + const finish = () => { + clearTimeout(timeout); + signal?.removeEventListener('abort', finish); + resolve(); + }; + + const timeout = setTimeout(finish, delayMs); + signal?.addEventListener('abort', finish, { once: true }); + }); +} + /** * Run async work items with a sliding-window concurrency pool. * Spawns up to `concurrency` workers that each grab the next diff --git a/packages/warden/src/utils/index.ts b/packages/warden/src/utils/index.ts index 4b57e832e..2c622b54b 100644 --- a/packages/warden/src/utils/index.ts +++ b/packages/warden/src/utils/index.ts @@ -1,4 +1,4 @@ -export { processInBatches, runPool, Semaphore } from './async.js'; +export { AsyncWorkQueue, processInBatches, runPool } from './async.js'; export { getVersion, getMajorVersion } from './version.js'; export { ExecError, diff --git a/skills/warden/references/cli-reference.md b/skills/warden/references/cli-reference.md index 7153007cd..3e5aa4e18 100644 --- a/skills/warden/references/cli-reference.md +++ b/skills/warden/references/cli-reference.md @@ -50,7 +50,7 @@ Ambiguous targets (no path separator, no extension) are resolved by checking if | `--fail-on ` | Exit with code 1 if findings >= severity | | `--report-on ` | Only show findings >= severity in output | | `--fix` | Automatically apply all suggested fixes | -| `--parallel ` | Max concurrent skill executions (default: 4) | +| `--parallel ` | Max concurrent hunk analyses across skills (default: 4) | | `--git` | Force ambiguous targets to be treated as git refs | | `--offline` | Use cached remote skills without network access | | `-q, --quiet` | Errors and final summary only | From ae9341d4f22ade9b7334719cb8b06895ce6d9981 Mon Sep 17 00:00:00 2001 From: Greg Pstrucha <875316+gricha@users.noreply.github.com> Date: Mon, 24 Aug 2026 11:39:21 -0700 Subject: [PATCH 2/2] fix(runner): Cancel sibling hunks on fatal errors Give every skill run a shared abort controller and cancel it when queued hunk execution throws. This stops in-flight work and prevents queued hunks from making provider calls after terminal failures. Co-Authored-By: GPT-5.6 Sol --- packages/warden/src/sdk/analyze.test.ts | 49 ++++++++++++++++++++++++- packages/warden/src/sdk/analyze.ts | 13 +++++-- 2 files changed, 57 insertions(+), 5 deletions(-) diff --git a/packages/warden/src/sdk/analyze.test.ts b/packages/warden/src/sdk/analyze.test.ts index d9ff7c5f3..3bfa2d156 100644 --- a/packages/warden/src/sdk/analyze.test.ts +++ b/packages/warden/src/sdk/analyze.test.ts @@ -5,9 +5,9 @@ import type { HunkWithContext } from '../diff/index.js'; import type { EventContext, Finding, UsageStats } from '../types/index.js'; import { analyzeFile, buildSourceSnippet, filterOutOfRangeFindings, runSkill } from './analyze.js'; import type { PreparedFile } from './types.js'; -import { getRuntime, type Runtime } from './runtimes/index.js'; +import { getRuntime, type Runtime, type SkillRunRequest } from './runtimes/index.js'; import { ProviderFailureCircuitBreaker } from './circuit-breaker.js'; -import { SkillRunnerError } from './errors.js'; +import { SkillRunnerError, WardenAuthenticationError } from './errors.js'; import { Sentry } from '../sentry.js'; import { startTracedSpan } from '../sentry-trace.js'; import { AsyncWorkQueue } from '../utils/index.js'; @@ -852,6 +852,51 @@ describe('runSkill', () => { vi.restoreAllMocks(); }); + it('cancels in-flight and queued hunks after an authentication failure', async () => { + let invocation = 0; + let markSecondStarted: () => void; + const secondStarted = new Promise((resolve) => { + markSecondStarted = resolve; + }); + const controllers = new Set(); + const runSkillMock = vi.fn(async (request: SkillRunRequest) => { + const controller = request.options.abortController; + expect(controller).toBeInstanceOf(AbortController); + controllers.add(controller!); + + const currentInvocation = invocation++; + if (currentInvocation === 0) { + await secondStarted; + throw new WardenAuthenticationError('bad credentials', { runtime: 'pi' }); + } + + markSecondStarted!(); + await new Promise((_resolve, reject) => { + controller!.signal.addEventListener('abort', () => reject(makeAbortError()), { once: true }); + }); + }); + vi.mocked(getRuntime).mockReturnValue({ + name: 'pi', + runSkill: runSkillMock, + runAuxiliary: vi.fn(), + runSynthesis: vi.fn(), + } as unknown as Runtime); + + await expect(runSkill( + { + name: 'security-review', + description: 'Security review.', + prompt: 'Return findings as JSON.', + }, + makeContextWithThreeHunks(), + { runtime: 'pi', concurrency: 2, verifyFindings: false }, + )).rejects.toBeInstanceOf(WardenAuthenticationError); + + expect(runSkillMock).toHaveBeenCalledTimes(2); + expect(controllers.size).toBe(1); + expect([...controllers][0]!.signal.aborted).toBe(true); + }); + it('reports all-extraction failures without authentication guidance', async () => { vi.mocked(getRuntime).mockReturnValue({ name: 'pi', diff --git a/packages/warden/src/sdk/analyze.ts b/packages/warden/src/sdk/analyze.ts index 8bb415608..0b2e6082c 100644 --- a/packages/warden/src/sdk/analyze.ts +++ b/packages/warden/src/sdk/analyze.ts @@ -844,7 +844,10 @@ export async function analyzeFile( }, }, async (span) => { - const { abortController } = options; + const abortController = options.abortController ?? new AbortController(); + const hunkOptions: SkillRunnerOptions = options.abortController + ? options + : { ...options, abortController }; const fileFindings: Finding[] = []; const fileUsage: UsageStats[] = []; const fileAuxiliaryUsage: AuxiliaryUsageEntry[] = []; @@ -883,11 +886,14 @@ export async function analyzeFile( skill, hunk, repoPath, - options, + hunkOptions, hunkCallbacks, prContext, span, - ); + ).catch((error: unknown) => { + abortController.abort(); + throw error; + }); const hunkDurationMs = Date.now() - hunkStartTime; attachElapsedTime(result.findings, callbacks?.skillStartTime); @@ -1013,6 +1019,7 @@ export async function runSkill( // This clone's identity scopes circuit-breaker provider diagnostics to this skill run. const scopedOptions: SkillRunnerOptions = { ...options, + abortController: options.abortController ?? new AbortController(), }; return Sentry.startSpan( {