fix(campaign): preserve caller cancellation

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.
 
 ---
 

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

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<void>((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<TestArtifact>((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)
+  })
 })

src/campaign/presets/run-optimization.ts

@@ -401,6 +401,7 @@ export async function runOptimization<TScenario extends Scenario, TArtifact>(
           coverage,
         }
       },
+      opts.signal,
     )
     for (const result of surfaceResults) {
       const { surface, surfaceHash: hash, campaign, coverage, label, rationale } = result

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<void>((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')

src/concurrency.ts

@@ -122,11 +122,13 @@ export async function mapConcurrent<T, R>(
   items: readonly T[],
   concurrency: number,
   map: (item: T, index: number, signal: AbortSignal) => Promise<R>,
+  signal?: AbortSignal,
 ): Promise<R[]> {
   return mapConcurrentRange({
     count: items.length,
     maxConcurrency: concurrency,
     label: 'mapConcurrent',
+    signal,
     map(index, signal) {
       return map(items[index]!, index, signal)
     },