diff --git a/CHANGELOG.md b/CHANGELOG.md index 125ef232..ff305181 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,7 +8,7 @@ All notable changes to `@tangle-network/agent-eval` and its sibling `agent-eval- ### Fixed -- Concurrent candidate campaigns now cancel active siblings after the first candidate failure. +- Concurrent candidate campaigns now cancel active siblings after the first candidate failure while preserving caller cancellation. --- diff --git a/src/analyst/benchmark-implementation.ts b/src/analyst/benchmark-implementation.ts index 513d2843..43b4e4d0 100644 --- a/src/analyst/benchmark-implementation.ts +++ b/src/analyst/benchmark-implementation.ts @@ -136,7 +136,7 @@ export const ANALYST_BENCHMARK_IMPLEMENTATION_FILES = Object.freeze([ ]) export const ANALYST_BENCHMARK_IMPLEMENTATION_SHA256 = - '4a960df94935a8177be12a64d20083510ca4afbfa050533a9356447f39c6f846' + '54cd9cc6d699dec0fced94f9850406adda8faf31a0280f6dbe2ff1bd31d795d5' export function analystBenchmarkImplementationDigest() { return ANALYST_BENCHMARK_IMPLEMENTATION_SHA256 diff --git a/src/campaign/presets/run-optimization.test.ts b/src/campaign/presets/run-optimization.test.ts index 38f11050..2ee60298 100644 --- a/src/campaign/presets/run-optimization.test.ts +++ b/src/campaign/presets/run-optimization.test.ts @@ -657,4 +657,56 @@ describe('runOptimization candidate concurrency', () => { expect(siblingSecondScenarioStarted).toBe(false) expect(dispatchedSurfaces).not.toContain('CANDIDATE-LATER') }) + + it('preserves caller cancellation inside an active candidate campaign', async () => { + const callerError = new Error('caller cancelled optimization') + const controller = new AbortController() + let candidateStarted!: () => void + const candidateReady = new Promise((resolve) => { + candidateStarted = resolve + }) + let candidateSignalAborted = false + + const pending = runOptimization({ + baselineSurface: 'BASELINE', + premeasuredBaseline: { + surfaceHash: surfaceHash('BASELINE'), + campaign: await measureBaseline(), + }, + scenarios, + dispatchWithSurface: async (surface, _scenario, ctx) => { + candidateStarted() + return new Promise((resolve, reject) => { + const timer = setTimeout(() => resolve({ surface: String(surface) }), 50) + ctx.signal.addEventListener( + 'abort', + () => { + clearTimeout(timer) + candidateSignalAborted = true + reject(ctx.signal.reason) + }, + { once: true }, + ) + }) + }, + judges: [qualityJudge], + proposer: proposer(), + populationSize: 1, + candidateConcurrency: 1, + maxConcurrency: 1, + maxGenerations: 1, + seed: 7, + signal: controller.signal, + runDir: '/parallel-candidates-caller-abort', + storage: inMemoryCampaignStorage(), + tracing: 'off', + expectUsage: 'off', + }) + + await candidateReady + controller.abort(callerError) + + await expect(pending).rejects.toBe(callerError) + expect(candidateSignalAborted).toBe(true) + }) }) diff --git a/src/campaign/presets/run-optimization.ts b/src/campaign/presets/run-optimization.ts index fc46eae2..c1b24e9a 100644 --- a/src/campaign/presets/run-optimization.ts +++ b/src/campaign/presets/run-optimization.ts @@ -401,6 +401,7 @@ export async function runOptimization( coverage, } }, + opts.signal, ) for (const result of surfaceResults) { const { surface, surfaceHash: hash, campaign, coverage, label, rationale } = result diff --git a/src/concurrency.test.ts b/src/concurrency.test.ts index 6b06ff67..7d878437 100644 --- a/src/concurrency.test.ts +++ b/src/concurrency.test.ts @@ -76,6 +76,30 @@ describe('mapConcurrent', () => { expect(started).toEqual([0, 1]) }) + it('forwards caller cancellation to active work and starts no later items', async () => { + const controller = new AbortController() + const reason = new Error('caller cancelled map') + const started: number[] = [] + const run = mapConcurrent( + [0, 1, 2], + 2, + async (value, _index, signal) => { + started.push(value) + await new Promise((resolve) => + signal.addEventListener('abort', () => resolve(), { once: true }), + ) + return value + }, + controller.signal, + ) + + await delay(1) + controller.abort(reason) + + await expect(run).rejects.toBe(reason) + expect(started).toEqual([0, 1]) + }) + it('validates the worker count and handles empty input', async () => { await expect(mapConcurrent([], 1, async () => 'unused')).resolves.toEqual([]) await expect(mapConcurrent([1], 0, async (value) => value)).rejects.toThrow('positive integer') diff --git a/src/concurrency.ts b/src/concurrency.ts index 6baa1d21..3395c4eb 100644 --- a/src/concurrency.ts +++ b/src/concurrency.ts @@ -122,11 +122,13 @@ export async function mapConcurrent( items: readonly T[], concurrency: number, map: (item: T, index: number, signal: AbortSignal) => Promise, + signal?: AbortSignal, ): Promise { return mapConcurrentRange({ count: items.length, maxConcurrency: concurrency, label: 'mapConcurrent', + signal, map(index, signal) { return map(items[index]!, index, signal) },