fix(core): settle a cancelled workflow at once, and size the window by usable CPUs

packages/core/src/agents/runtime/workflow-orchestrator.test.ts

@@ -2468,7 +2468,7 @@ describe('WorkflowOrchestrator P2 — parallel() / pipeline() / caps', () => {
         args: undefined,
       });
       // 50 thunks >> window, so the window fully fills: peak === cap.
-      const cap = Math.max(1, Math.min(16, os.cpus().length - 2));
+      const cap = Math.max(2, Math.min(16, os.availableParallelism() - 2));
       expect(peak).toBe(cap);
     });
   });
@@ -2541,7 +2541,7 @@ describe('WorkflowOrchestrator P2 — parallel() / pipeline() / caps', () => {
         );`,
         args: undefined,
       });
-      const cap = Math.max(1, Math.min(16, os.cpus().length - 2));
+      const cap = Math.max(2, Math.min(16, os.availableParallelism() - 2));
       expect(peak).toBe(cap);
     });
 
@@ -2836,18 +2836,33 @@ describe('WorkflowOrchestrator P2 — parallel() / pipeline() / caps', () => {
       }
     });
 
-    it('resolveConcurrencyLimit honors a valid override and clamps the cpu default to [1,16]', () => {
+    it('resolveConcurrencyLimit honors a valid override and clamps the cpu default to [2,16]', () => {
       expect(
         resolveConcurrencyLimit({ QWEN_CODE_MAX_WORKFLOW_CONCURRENCY: '4' }),
       ).toBe(4);
-      // invalid → cpu-derived default, always within [1, 16]
+      // invalid → cpu-derived default, always within [2, 16]
       const fallback = resolveConcurrencyLimit({
         QWEN_CODE_MAX_WORKFLOW_CONCURRENCY: '-1',
       });
-      expect(fallback).toBeGreaterThanOrEqual(1);
+      expect(fallback).toBeGreaterThanOrEqual(2);
       expect(fallback).toBeLessThanOrEqual(16);
     });
 
+    // The default reads `availableParallelism()`, which honours the CPU
+    // affinity mask and container limits, where `os.cpus()` reports the
+    // host and can return an empty array. The floor is 2, not 1: a window
+    // of 1 turns every `parallel()` into a sequence on a small machine.
+    it('resolveConcurrencyLimit derives the default from availableParallelism, floored at 2', () => {
+      const at = (parallelism: number) =>
+        resolveConcurrencyLimit({}, () => parallelism);
+      expect(at(0)).toBe(2);
+      expect(at(1)).toBe(2);
+      expect(at(3)).toBe(2);
+      expect(at(6)).toBe(4);
+      expect(at(18)).toBe(16);
+      expect(at(64)).toBe(16);
+    });
+
     // PR #4947 R1 T4 (wenshao): an env override above the hard ceiling must
     // be clamped, not honored — a single Node process running 999999
     // concurrent LLM calls would OOM long before saturating the model.

packages/core/src/agents/runtime/workflow-orchestrator.ts

@@ -124,15 +124,17 @@ export const HARD_MAX_CONCURRENCY_CEILING = 64;
 
 /**
  * Maximum agents in flight at once within a single run, shared across all
- * `parallel()` / `pipeline()` calls. `min(16, cpus-2)` mirrors upstream;
- * `max(1, …)` guards 1–2 core machines where `cpus-2 <= 0` would otherwise
- * produce a deadlocking limit. `QWEN_CODE_MAX_WORKFLOW_CONCURRENCY` overrides
- * the computed value with an explicit integer in `[1, HARD_MAX_CONCURRENCY_CEILING]`;
- * an invalid override falls back to the cpu-derived default with a debug
- * warning, and an over-ceiling override is clamped.
+ * `parallel()` / `pipeline()` calls. `min(16, availableParallelism()-2)`
+ * mirrors upstream; `max(2, …)` floors small machines at 2 — a window of 1
+ * would serialize every `parallel()` and silently defeat the point of a
+ * fan-out. `QWEN_CODE_MAX_WORKFLOW_CONCURRENCY` overrides the computed value
+ * with an explicit integer in `[1, HARD_MAX_CONCURRENCY_CEILING]`; an invalid
+ * override falls back to the cpu-derived default with a debug warning, and an
+ * over-ceiling override is clamped.
  */
 export function resolveConcurrencyLimit(
   env: Record<string, string | undefined> = process.env,
+  availableParallelism: () => number = os.availableParallelism,
 ): number {
   const raw = env[MAX_WORKFLOW_CONCURRENCY_ENV];
   if (raw !== undefined && raw.trim() !== '') {
@@ -154,7 +156,12 @@ export function resolveConcurrencyLimit(
         `using cpu-derived default`,
     );
   }
-  return Math.max(1, Math.min(16, os.cpus().length - 2));
+  // `availableParallelism()` honours the process's CPU affinity mask and
+  // container CPU limits; `os.cpus()` reports the host and can return an
+  // empty array in some sandboxes, which used to make every run serial.
+  // Floor of 2: a window of 1 turns `parallel()` into a sequence and
+  // silently defeats the point of a fan-out on a small machine.
+  return Math.max(2, Math.min(16, availableParallelism() - 2));
 }
 
 /**

packages/core/src/agents/runtime/workflow-runner.ts

@@ -320,10 +320,26 @@ export class WorkflowRunner {
           const message = extractErrorMessage(error);
           if (entry && details?.meta && !entry.meta) entry.meta = details.meta;
           if (details?.logs) registry?.setRecentLogs(runId, details.logs);
+          // Mirror of the guard on the success path. When the entry was
+          // settled terminal from outside — the dialog's cancel, or the
+          // approval contingency's fail — the abort that follows is what
+          // makes the sandbox reject, so the rejection arriving here is a
+          // consequence of that settlement, not a new fact about the run.
+          // Report the entry's state and its own message, not the
+          // rejection's.
+          if (entry && isTerminalWorkflowStatus(entry.status)) {
+            return {
+              ok: false,
+              message:
+                entry.status === 'cancelled'
+                  ? 'Workflow run cancelled.'
+                  : (entry.error ?? message),
+              details,
+            };
+          }
           if (
             callerWasAbortedBeforeStart ||
-            (!runInBackground && options.signal.aborted) ||
-            entry?.status === 'cancelled'
+            (!runInBackground && options.signal.aborted)
           ) {
             registry?.cancel(runId, Date.now());
           } else {

packages/core/src/agents/runtime/workflow-sandbox.test.ts

@@ -1133,12 +1133,12 @@ describe('createWorkflowSandbox security', () => {
     await expect(run).rejects.toThrow(/exceeded 100 ms of active time/);
   });
 
-  it('re-arms the pause-suspended watchdog on abort so a cancelled paused run still settles', async () => {
+  it('settles a cancelled paused run at once, not on the banked wall-clock remainder', async () => {
     // Cancelling a paused run aborts the controller, but abortPending()
-    // emits no scheduler transition — a pause-suspended watchdog that
-    // never re-armed would leave a script hung in ungated code pending
-    // forever (no settlement, snapshot, or telemetry). The abort must
-    // re-arm the banked remainder.
+    // emits no scheduler transition. Before the abort arm, settlement
+    // depended on the watchdog re-arming with its banked remainder — the
+    // user watched a cancelled run refuse to end for up to that long. Now
+    // the abort settles the run immediately; the re-arm stays as a backstop.
     vi.useFakeTimers();
     const scheduler = new WorkflowDispatchScheduler(1);
     const abortOnTimeout = new AbortController();
@@ -1168,10 +1168,9 @@ describe('createWorkflowSandbox security', () => {
       expect(scheduler.snapshot().state).toBe('paused');
 
       abortOnTimeout.abort();
-      await vi.advanceTimersByTimeAsync(20);
-      expect(settled).toBe(false);
-      await vi.advanceTimersByTimeAsync(100);
-      await expect(run).rejects.toThrow(/exceeded 200 ms of active time/);
+      await vi.advanceTimersByTimeAsync(0);
+      expect(settled).toBe(true);
+      await expect(run).rejects.toThrow(/aborted \(cancelled\)/);
     } finally {
       abortOnTimeout.abort();
       await vi.runAllTimersAsync();
@@ -1180,12 +1179,11 @@ describe('createWorkflowSandbox security', () => {
     }
   });
 
-  it('keeps the watchdog armed when a post-abort drain lands paused', async () => {
+  it('settles a run cancelled mid-pausing before the post-abort drain lands', async () => {
     // An in-flight dispatch that settles AFTER the abort still lands the
-    // scheduler's `pausing` → `paused` transition (pump's finally). The
-    // watchdog must not re-suspend on that post-abort transition, or a
-    // script that catches the abort and hangs in ungated code is
-    // orphaned again.
+    // scheduler's `pausing` → `paused` transition (pump's finally). The run
+    // must already be settled by then: a script that catches the abort and
+    // hangs in ungated code used to be orphaned until the wall clock.
     const scheduler = new WorkflowDispatchScheduler(1);
     const abortOnTimeout = new AbortController();
     let finishDispatch: ((value: string) => void) | undefined;
@@ -1213,12 +1211,12 @@ describe('createWorkflowSandbox security', () => {
     expect(scheduler.snapshot().state).toBe('pausing');
 
     abortOnTimeout.abort();
+    await expect(run).rejects.toThrow(/aborted \(cancelled\)/);
+
+    // The drain still completes its transition afterwards; nothing about a
+    // settled run is disturbed by it.
     finishDispatch?.('late');
     await vi.waitFor(() => expect(scheduler.snapshot().state).toBe('paused'));
-
-    // The post-abort `paused` transition must not re-suspend the
-    // watchdog — the hung script still settles on the banked remainder.
-    await expect(run).rejects.toThrow(/exceeded 150 ms of active time/);
   });
 
   it('suspends the watchdog of a sandbox created while the scheduler is already paused', async () => {
@@ -1254,28 +1252,59 @@ describe('createWorkflowSandbox security', () => {
     await expect(run).rejects.toThrow(/exceeded 100 ms of active time/);
   });
 
-  it('keeps the seeded watchdog armed when the shared signal is already aborted', async () => {
+  it('does not run a script whose signal was already aborted before run()', async () => {
     // Production shares one controller between the registry (which
     // pre-registers the run before run() resolves) and the sandbox, so a
     // user cancel can land before run() begins: the scheduler is already
-    // paused AND the signal is already aborted. The abort re-arm listener
-    // is dead on an already-aborted signal and no scheduler transition
-    // can follow, so only the seed guard's !aborted() check keeps the
-    // watchdog armed — deleting it suspends the newborn watchdog and the
-    // hung run never settles.
+    // paused AND the signal is already aborted. The run must settle as
+    // cancelled without executing a line of model-authored code.
     const abortOnTimeout = new AbortController();
     const scheduler = new WorkflowDispatchScheduler(1, abortOnTimeout.signal);
     expect(scheduler.pause()).toBe(true);
     abortOnTimeout.abort();
+    const dispatch = vi.fn(async () => 'ignored');
     const sandbox = createWorkflowSandbox({
       args: undefined,
-      dispatch: async () => 'ignored',
+      dispatch,
       maxWallClockMs: 100,
       scheduler,
       abortOnTimeout,
     });
-    await expect(sandbox.run(`return new Promise(() => {});`)).rejects.toThrow(
-      /exceeded 100 ms of active time/,
+    await expect(
+      sandbox.run(`await agent('never'); return new Promise(() => {});`),
+    ).rejects.toThrow(/aborted \(cancelled\)/);
+    expect(dispatch).not.toHaveBeenCalled();
+  });
+
+  it('cancellation settles the run at once while the script still runs its own finally', async () => {
+    // The host-side run settles immediately; the script's promise is left
+    // to finish on its own. Its dispatches reject on the aborted signal, so
+    // a `finally` inside the script still executes — cleanup the author
+    // wrote is not skipped just because the user stopped waiting.
+    const controller = new AbortController();
+    const sandbox = createWorkflowSandbox({
+      args: undefined,
+      abortOnTimeout: controller,
+      dispatch: () =>
+        new Promise<string>((_, reject) => {
+          controller.signal.addEventListener(
+            'abort',
+            () => reject(new Error('dispatch aborted')),
+            { once: true },
+          );
+        }),
+    });
+    const run = sandbox.run(`
+      try { await agent('a'); }
+      finally { log('cleanup ran'); }
+    `);
+    await new Promise((resolve) => setTimeout(resolve, 10));
+    controller.abort();
+    await expect(run).rejects.toThrow(/aborted \(cancelled\)/);
+    await vi.waitFor(() =>
+      expect(sandbox.getLogs().some((l) => l.includes('cleanup ran'))).toBe(
+        true,
+      ),
     );
   });
 
@@ -2111,7 +2140,10 @@ describe('createWorkflowSandbox primitives', () => {
         'Workflow subagent x did not complete (terminate mode: CANCELLED).',
       ),
     );
-    await runPromise;
+    // The abort arm settles the run as cancelled even though the script had
+    // already returned — cancel wins a same-tick race, matching the runner,
+    // which reports a cancelled registry entry over an ok outcome.
+    await expect(runPromise).rejects.toThrow(/aborted \(cancelled\)/);
     expect(sandbox.getLogs()).toEqual([]);
   });
 

packages/core/src/agents/runtime/workflow-sandbox.ts

@@ -245,6 +245,17 @@ export function compileWorkflowScript(scriptSource: string): {
   return { script, meta };
 }
 
+/**
+ * The error a cancelled run settles with. Named `AbortError` so every
+ * `isAbortError` check on the way up classifies it as a cancellation rather
+ * than a failure.
+ */
+function workflowCancelledError(): Error {
+  const error = new Error('Workflow run was aborted (cancelled).');
+  error.name = 'AbortError';
+  return error;
+}
+
 /** Longest source line rendered in a compile-failure message. */
 const COMPILE_ERROR_LINE_WIDTH = 80;
 
@@ -1860,6 +1871,8 @@ export function createWorkflowSandbox(opts: SandboxOptions): WorkflowSandbox {
       let watchdog: WallClockWatchdog | undefined;
       let stopWatchingState: (() => void) | undefined;
       let rearmWatchdogOnAbort: (() => void) | undefined;
+      let settleOnAbort: (() => void) | undefined;
+      let wallClockFired = false;
       process.on('unhandledRejection', adoptionEscapeHook);
       try {
         // P4: extract `export const meta = {...}` once before the body runs.
@@ -1874,6 +1887,12 @@ export function createWorkflowSandbox(opts: SandboxOptions): WorkflowSandbox {
         // which is the failure the gate exists to prevent.
         const { script, meta } = compileWorkflowScript(scriptSource);
         extractedMeta = meta;
+        // A run cancelled before its script started must not start it: the
+        // registry pre-registers the run and shares this controller, so a
+        // user cancel can land before `run()` is entered.
+        if (opts.abortOnTimeout?.signal.aborted) {
+          throw workflowCancelledError();
+        }
         // 30s sync wall-clock cap inside vm — covers `while(true){}` style
         // synchronous loops only. Once the IIFE hits its first `await`,
         // `runInContext` returns and this timer is disarmed.
@@ -1892,7 +1911,10 @@ export function createWorkflowSandbox(opts: SandboxOptions): WorkflowSandbox {
             // T40 (PR #4732 R4): abort linked controller BEFORE rejecting so
             // in-flight subagents see the cancellation and stop. Order
             // matters: rejecting first then aborting would race the
-            // caller's finally block.
+            // caller's finally block. The abort arm below must let this
+            // abort through untouched — a timeout is not a cancellation,
+            // and it has to be reported as the timeout it is.
+            wallClockFired = true;
             opts.abortOnTimeout?.abort();
             reject(
               new Error(
@@ -1929,27 +1951,50 @@ export function createWorkflowSandbox(opts: SandboxOptions): WorkflowSandbox {
         if (opts.scheduler?.snapshot().state === 'paused' && !aborted()) {
           watchdog?.pause();
         }
-        // Cancellation must settle the run even when the script hangs in
-        // ungated code: `registry.cancel()` aborts this controller, but
-        // `abortPending()` emits no state transition, so a pause-suspended
-        // watchdog would never re-arm and the race below has no abort arm
-        // — the hung run would never reach its settlement `finally`
-        // (snapshot, telemetry, and handle release all skipped). Re-arm
-        // with the banked remainder on abort to restore the bound.
+        // Belt to the abort arm's braces below: `registry.cancel()` aborts
+        // this controller, but `abortPending()` emits no state transition,
+        // so a pause-suspended watchdog would otherwise stay suspended.
+        // Re-arm it with the banked remainder so the wall clock keeps
+        // bounding the run independently of the abort arm.
         rearmWatchdogOnAbort = (): void => watchdog?.resume();
         opts.abortOnTimeout?.signal.addEventListener(
           'abort',
           rearmWatchdogOnAbort,
           { once: true },
         );
-        return await Promise.race([result, timeoutPromise]);
+        // Cancellation settles the run now, not when the wall clock runs
+        // out. Without this arm a script that is not currently blocked on a
+        // dispatch — sitting in ungated `await`s, or simply hung — keeps the
+        // run open until the banked remainder of the clock expires, and the
+        // user watches a cancelled run refuse to end. The script's own
+        // promise is left to settle by itself: its dispatches see the
+        // aborted signal and reject, so its `finally` blocks still run, and
+        // `Promise.race` keeps a handler attached so that later rejection
+        // is never an unhandled one.
+        const abortPromise = new Promise<never>((_, reject) => {
+          const signal = opts.abortOnTimeout?.signal;
+          if (!signal) return;
+          settleOnAbort = (): void => {
+            // The watchdog aborts the controller itself on the way to
+            // rejecting with the timeout; that abort is not a cancellation.
+            if (!wallClockFired) reject(workflowCancelledError());
+          };
+          signal.addEventListener('abort', settleOnAbort, { once: true });
+        });
+        return await Promise.race([result, timeoutPromise, abortPromise]);
       } finally {
         if (rearmWatchdogOnAbort) {
           opts.abortOnTimeout?.signal.removeEventListener(
             'abort',
             rearmWatchdogOnAbort,
           );
         }
+        if (settleOnAbort) {
+          opts.abortOnTimeout?.signal.removeEventListener(
+            'abort',
+            settleOnAbort,
+          );
+        }
         stopWatchingState?.();
         watchdog?.stop();
         // R11-10: the flush shares the finally that wraps the ENTIRE