diff --git a/packages/core/src/agents/runtime/workflow-orchestrator.test.ts b/packages/core/src/agents/runtime/workflow-orchestrator.test.ts
index f94df40d77b..b80357eb8ba 100644
--- a/packages/core/src/agents/runtime/workflow-orchestrator.test.ts
+++ b/packages/core/src/agents/runtime/workflow-orchestrator.test.ts
@@ -7,10 +7,14 @@
 // T7 (PR #4732 R1): the `vi as vitest` alias diverges from every other
 // test file in the repo. Use `vi` directly.
 import { describe, it, expect, beforeEach, vi } from 'vitest';
+import * as os from 'node:os';
 import {
   WorkflowOrchestrator,
   WorkflowExecutionError,
   createProductionDispatch,
+  DEFAULT_MAX_AGENTS_PER_RUN,
+  resolveMaxAgentsPerRun,
+  resolveConcurrencyLimit,
 } from './workflow-orchestrator.js';
 import type { Config } from '../../config/config.js';
 
@@ -292,3 +296,428 @@ describe('WorkflowOrchestrator failure-context preservation', () => {
     expect(wfErr.logs).toEqual(['starting', 'about to fail']);
   });
 });
+
+describe('WorkflowOrchestrator P2 — parallel() / pipeline() / caps', () => {
+  describe('parallel()', () => {
+    it('resolves all thunks to a position-aligned array', async () => {
+      const orchestrator = new WorkflowOrchestrator(
+        async (prompt) => `r:${prompt}`,
+      );
+      const outcome = await orchestrator.run({
+        script: `return await parallel([
+          () => agent("a"),
+          () => agent("b"),
+          () => agent("c"),
+        ]);`,
+        args: undefined,
+      });
+      expect(outcome.result).toEqual(['r:a', 'r:b', 'r:c']);
+    });
+
+    it('errors-as-data: a thunk that throws becomes null at its index, others unaffected', async () => {
+      const orchestrator = new WorkflowOrchestrator(
+        async (prompt) => `r:${prompt}`,
+      );
+      const outcome = await orchestrator.run({
+        script: `return await parallel([
+          () => agent("a"),
+          () => { throw new Error("boom"); },
+          () => agent("c"),
+        ]);`,
+        args: undefined,
+      });
+      expect(outcome.result).toEqual(['r:a', null, 'r:c']);
+    });
+
+    it('rejects on a non-function element (eager promise instead of thunk)', async () => {
+      const orchestrator = new WorkflowOrchestrator(async () => 'unused');
+      await expect(
+        orchestrator.run({
+          script: `return await parallel([agent("a")]);`,
+          args: undefined,
+        }),
+      ).rejects.toThrow(/array of functions/);
+    });
+
+    // EAD-1 (P2 self-review): a thunk that resolves to a non-JSON-serializable
+    // value (BigInt / circular) must become null at its index — NOT crash the
+    // whole batch. The in-realm revival is per-element, so one bad slot cannot
+    // destroy its siblings (errors-as-data holds for return values too).
+    it('a thunk returning a non-serializable value becomes null without crashing siblings', async () => {
+      const orchestrator = new WorkflowOrchestrator(async () => 'unused');
+      const outcome = await orchestrator.run({
+        script: `return await parallel([
+          () => "a",
+          () => 1n,
+          () => "c",
+          () => { const o = {}; o.self = o; return o; },
+        ]);`,
+        args: undefined,
+      });
+      expect(outcome.result).toEqual(['a', null, 'c', null]);
+    });
+
+    it('caps concurrent agents within a fan-out to the shared per-run window', async () => {
+      let inFlight = 0;
+      let peak = 0;
+      const orchestrator = new WorkflowOrchestrator(async () => {
+        inFlight++;
+        peak = Math.max(peak, inFlight);
+        await new Promise((r) => setTimeout(r, 5));
+        inFlight--;
+        return 'ok';
+      });
+      await orchestrator.run({
+        script: `return await parallel(
+          Array.from({ length: 50 }, () => () => agent("x"))
+        );`,
+        args: undefined,
+      });
+      // 50 thunks >> window, so the window fully fills: peak === cap.
+      const cap = Math.max(1, Math.min(16, os.cpus().length - 2));
+      expect(peak).toBe(cap);
+    });
+  });
+
+  describe('pipeline()', () => {
+    it('runs each item through the stages; first stage receives (item, item, idx)', async () => {
+      const orchestrator = new WorkflowOrchestrator(async () => 'unused');
+      const outcome = await orchestrator.run({
+        script: `return await pipeline([10, 20],
+          (prev, item, idx) => prev + "|" + item + "|" + idx,
+          (prev) => "S2(" + prev + ")",
+        );`,
+        args: undefined,
+      });
+      expect(outcome.result).toEqual(['S2(10|10|0)', 'S2(20|20|1)']);
+    });
+
+    it('a stage returning null drops that item to null and skips remaining stages', async () => {
+      const orchestrator = new WorkflowOrchestrator(async () => 'unused');
+      const outcome = await orchestrator.run({
+        script: `return await pipeline([1, 2, 3],
+          (x) => (x === 2 ? null : x),
+          (x) => x * 100,
+        );`,
+        args: undefined,
+      });
+      expect(outcome.result).toEqual([100, null, 300]);
+    });
+
+    it('a stage that throws drops that item to null (errors-as-data), others unaffected', async () => {
+      const orchestrator = new WorkflowOrchestrator(async () => 'unused');
+      const outcome = await orchestrator.run({
+        script: `return await pipeline([1, 2, 3],
+          (x) => { if (x === 2) throw new Error("bad"); return x; },
+          (x) => x * 100,
+        );`,
+        args: undefined,
+      });
+      expect(outcome.result).toEqual([100, null, 300]);
+    });
+
+    it('rejects when a stage is not a function', async () => {
+      const orchestrator = new WorkflowOrchestrator(async () => 'unused');
+      await expect(
+        orchestrator.run({
+          script: `return await pipeline([1, 2], "not a function");`,
+          args: undefined,
+        }),
+      ).rejects.toThrow(/stages must be functions/);
+    });
+
+    // TST-1 (P2 self-review): pipeline must share the SAME per-run window as
+    // parallel — a pipeline impl that gave itself a separate (or no) limiter
+    // would let concurrency exceed the cap. Drive 50 item-chains, each calling
+    // one agent per stage, and assert peak in-flight === cap.
+    it('caps concurrent agents across a pipeline fan-out (shares the run window)', async () => {
+      let inFlight = 0;
+      let peak = 0;
+      const orchestrator = new WorkflowOrchestrator(async () => {
+        inFlight++;
+        peak = Math.max(peak, inFlight);
+        await new Promise((r) => setTimeout(r, 5));
+        inFlight--;
+        return 'ok';
+      });
+      await orchestrator.run({
+        script: `return await pipeline(
+          Array.from({ length: 50 }, (_, i) => i),
+          (x) => agent("s1-" + x),
+        );`,
+        args: undefined,
+      });
+      const cap = Math.max(1, Math.min(16, os.cpus().length - 2));
+      expect(peak).toBe(cap);
+    });
+
+    // TST-2 (P2 self-review): pipeline is parallel-of-chains — STAGGERED, with
+    // NO inter-stage barrier. Item 0's stage-2 dispatch must be able to fire
+    // WHILE item 1's stage-1 dispatch is still pending. We assert this
+    // deterministically via a release gate: item 1's stage 1 blocks until
+    // item 0 reaches stage 2 and releases the gate. A barrier impl
+    // (all items must clear stage N before any enters stage N+1) cannot
+    // satisfy this — it would wait for item 1's stage 1 to finish first,
+    // creating a circular wait that the vitest timeout would catch.
+    //
+    // PR #4947 R2 T6 (DragonnZhang): the previous version of this test used
+    // an elapsed-time threshold (50ms vs 120ms) which fails deterministically
+    // on a 3-core macOS-14 CI runner because the cpu-derived concurrency
+    // limit becomes 1 — FIFO then forces all stage-1 dispatches to settle
+    // before any stage-2 starts, breaking the timing assumption. Force the
+    // limit to 2 AND use a release gate so the assertion is timing-free.
+    it('is staggered with no inter-stage barrier (item A reaches stage 2 while item B is still in stage 1)', async () => {
+      const envPrev = process.env['QWEN_CODE_MAX_WORKFLOW_CONCURRENCY'];
+      process.env['QWEN_CODE_MAX_WORKFLOW_CONCURRENCY'] = '2';
+      try {
+        let releaseItem1Stage1: () => void = () => {};
+        const item1Stage1Gate = new Promise<void>((resolve) => {
+          releaseItem1Stage1 = resolve;
+        });
+        let item0ReachedStage2 = false;
+
+        const orchestrator = new WorkflowOrchestrator(async (prompt) => {
+          if (prompt === 's1-0') return 'ok'; // item 0's stage 1: fast
+          if (prompt === 's1-1') {
+            // item 1's stage 1: BLOCKS until item 0 reaches stage 2.
+            // A staggered impl lets item 0 advance to stage 2 while this is
+            // still in flight. A barrier impl would hang here waiting for
+            // item 0's stage 2 — but item 0's stage 2 cannot start because
+            // the barrier is waiting for THIS to finish. Mutual wait =
+            // vitest timeout = test fails for a barrier impl.
+            await item1Stage1Gate;
+            return 'ok';
+          }
+          if (prompt === 's2-0') {
+            item0ReachedStage2 = true;
+            releaseItem1Stage1();
+            return 'ok';
+          }
+          if (prompt === 's2-1') return 'ok';
+          throw new Error(`unexpected prompt ${prompt}`);
+        });
+
+        await orchestrator.run({
+          script: `return await pipeline([0, 1],
+            (prev, item) => agent("s1-" + item),
+            (prev, item) => agent("s2-" + item),
+          );`,
+          args: undefined,
+        });
+
+        expect(item0ReachedStage2).toBe(true);
+      } finally {
+        if (envPrev === undefined)
+          delete process.env['QWEN_CODE_MAX_WORKFLOW_CONCURRENCY'];
+        else process.env['QWEN_CODE_MAX_WORKFLOW_CONCURRENCY'] = envPrev;
+      }
+    }, 10_000);
+  });
+
+  describe('1000-agent cap', () => {
+    it('the 1001st sequential agent() call throws the cap error', async () => {
+      const orchestrator = new WorkflowOrchestrator(async () => 'ok');
+      await expect(
+        orchestrator.run({
+          script: `for (let i = 0; i < ${DEFAULT_MAX_AGENTS_PER_RUN + 1}; i++) {
+            await agent("x");
+          }
+          return "done";`,
+          args: undefined,
+        }),
+      ).rejects.toThrow(
+        new RegExp(`${DEFAULT_MAX_AGENTS_PER_RUN} agent\\(\\) calls per run`),
+      );
+    });
+
+    it('the cap counts agents launched via parallel() — a fan-out cannot bypass it', async () => {
+      const orchestrator = new WorkflowOrchestrator(async () => 'ok');
+      const outcome = await orchestrator.run({
+        script: `return await parallel(
+          Array.from({ length: ${DEFAULT_MAX_AGENTS_PER_RUN + 1} }, () => () => agent("x"))
+        );`,
+        args: undefined,
+      });
+      const arr = outcome.result as Array<string | null>;
+      // Exactly 1000 dispatches succeed; the one over the cap becomes null.
+      expect(arr.filter((v) => v === 'ok')).toHaveLength(
+        DEFAULT_MAX_AGENTS_PER_RUN,
+      );
+      expect(arr.filter((v) => v === null)).toHaveLength(1);
+    });
+  });
+
+  describe('abort', () => {
+    it('parallel() rejects (not silent nulls) when the run is aborted', async () => {
+      const ac = new AbortController();
+      ac.abort();
+      const orchestrator = new WorkflowOrchestrator(async () => 'ok');
+      await expect(
+        orchestrator.run({
+          script: `return await parallel([() => agent("a"), () => agent("b")]);`,
+          args: undefined,
+          abortOnTimeout: ac,
+        }),
+      ).rejects.toThrow(/abort/i);
+    });
+
+    // TST-3 (P2 self-review): the pre-aborted case above only exercises the
+    // fast-path. Abort MID-FLIGHT — after dispatches have already started —
+    // and confirm parallel() rejects rather than resolving with a silent array
+    // of nulls (which would let an aborted/timed-out workflow continue).
+    it('parallel() rejects when aborted MID-FLIGHT (after dispatches started)', async () => {
+      const ac = new AbortController();
+      let dispatched = 0;
+      const orchestrator = new WorkflowOrchestrator(async () => {
+        dispatched++;
+        await new Promise((r) => setTimeout(r, 40));
+        return 'ok';
+      });
+      const p = orchestrator.run({
+        script: `return await parallel(
+          Array.from({ length: 6 }, () => () => agent("x"))
+        );`,
+        args: undefined,
+        abortOnTimeout: ac,
+      });
+      // Abort once at least one dispatch is in flight.
+      setTimeout(() => ac.abort(), 10);
+      await expect(p).rejects.toThrow(/abort/i);
+      expect(dispatched).toBeGreaterThan(0);
+    }, 10_000);
+  });
+
+  describe('nested fan-out (shared-window re-entrancy)', () => {
+    // F1 (P2 review round 1): the concurrency limiter throttles AGENT
+    // DISPATCHES, not orchestration thunks. `pipeline([items], item =>
+    // parallel([...]))` is a canonical pattern (it's in the upstream
+    // /deep-research workflow). If the limiter sat at the thunk level, the
+    // outer pipeline chains would each hold a slot while awaiting an inner
+    // parallel(), and on a low concurrency limit every slot is held by a
+    // stalled outer thunk → the inner agent() calls can never acquire a slot
+    // → unrecoverable deadlock (silent hang until the 30-min wall clock).
+    // Forcing the window to 1 makes the worst case deterministic.
+    it('a parallel() inside a pipeline() stage does not deadlock at concurrency=1', async () => {
+      const prev = process.env['QWEN_CODE_MAX_WORKFLOW_CONCURRENCY'];
+      process.env['QWEN_CODE_MAX_WORKFLOW_CONCURRENCY'] = '1';
+      try {
+        const orchestrator = new WorkflowOrchestrator(async (p) => `r:${p}`);
+        const outcome = await orchestrator.run({
+          script: `return await pipeline([0, 1, 2],
+            (prev, item) => parallel([() => agent("a" + item)]),
+          );`,
+          args: undefined,
+        });
+        expect(outcome.result).toEqual([['r:a0'], ['r:a1'], ['r:a2']]);
+      } finally {
+        if (prev === undefined)
+          delete process.env['QWEN_CODE_MAX_WORKFLOW_CONCURRENCY'];
+        else process.env['QWEN_CODE_MAX_WORKFLOW_CONCURRENCY'] = prev;
+      }
+    }, 15_000);
+
+    it('parallel() of parallel() does not deadlock at concurrency=1', async () => {
+      const prev = process.env['QWEN_CODE_MAX_WORKFLOW_CONCURRENCY'];
+      process.env['QWEN_CODE_MAX_WORKFLOW_CONCURRENCY'] = '1';
+      try {
+        const orchestrator = new WorkflowOrchestrator(async (p) => `r:${p}`);
+        const outcome = await orchestrator.run({
+          script: `return await parallel([
+            () => parallel([() => agent("x")]),
+            () => parallel([() => agent("y")]),
+          ]);`,
+          args: undefined,
+        });
+        expect(outcome.result).toEqual([['r:x'], ['r:y']]);
+      } finally {
+        if (prev === undefined)
+          delete process.env['QWEN_CODE_MAX_WORKFLOW_CONCURRENCY'];
+        else process.env['QWEN_CODE_MAX_WORKFLOW_CONCURRENCY'] = prev;
+      }
+    }, 15_000);
+  });
+
+  describe('env-overridable caps', () => {
+    it('resolveMaxAgentsPerRun defaults to 1000 and honors a valid override', () => {
+      expect(resolveMaxAgentsPerRun({})).toBe(DEFAULT_MAX_AGENTS_PER_RUN);
+      expect(
+        resolveMaxAgentsPerRun({ QWEN_CODE_MAX_WORKFLOW_AGENTS: '50' }),
+      ).toBe(50);
+    });
+
+    it('resolveMaxAgentsPerRun rejects a non-integer / <1 override and falls back', () => {
+      expect(
+        resolveMaxAgentsPerRun({ QWEN_CODE_MAX_WORKFLOW_AGENTS: '0' }),
+      ).toBe(DEFAULT_MAX_AGENTS_PER_RUN);
+      expect(
+        resolveMaxAgentsPerRun({ QWEN_CODE_MAX_WORKFLOW_AGENTS: 'abc' }),
+      ).toBe(DEFAULT_MAX_AGENTS_PER_RUN);
+      expect(
+        resolveMaxAgentsPerRun({ QWEN_CODE_MAX_WORKFLOW_AGENTS: '2.5' }),
+      ).toBe(DEFAULT_MAX_AGENTS_PER_RUN);
+    });
+
+    // PR #4947 R1 T4 (wenshao): an env override above the hard ceiling must
+    // be clamped, not honored — protects operators from a fat-finger
+    // QWEN_CODE_MAX_WORKFLOW_AGENTS=999999999 silently uncapping the run.
+    it('resolveMaxAgentsPerRun clamps an over-ceiling override to the hard maximum', () => {
+      expect(
+        resolveMaxAgentsPerRun({
+          QWEN_CODE_MAX_WORKFLOW_AGENTS: '999999999',
+        }),
+      ).toBe(10_000);
+      // Just under the ceiling is preserved.
+      expect(
+        resolveMaxAgentsPerRun({ QWEN_CODE_MAX_WORKFLOW_AGENTS: '9999' }),
+      ).toBe(9999);
+    });
+
+    it('resolveConcurrencyLimit honors a valid override and clamps the cpu default to [1,16]', () => {
+      expect(
+        resolveConcurrencyLimit({ QWEN_CODE_MAX_WORKFLOW_CONCURRENCY: '4' }),
+      ).toBe(4);
+      // invalid → cpu-derived default, always within [1, 16]
+      const fallback = resolveConcurrencyLimit({
+        QWEN_CODE_MAX_WORKFLOW_CONCURRENCY: '-1',
+      });
+      expect(fallback).toBeGreaterThanOrEqual(1);
+      expect(fallback).toBeLessThanOrEqual(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.
+    it('resolveConcurrencyLimit clamps an over-ceiling override to the hard maximum', () => {
+      expect(
+        resolveConcurrencyLimit({
+          QWEN_CODE_MAX_WORKFLOW_CONCURRENCY: '999999',
+        }),
+      ).toBe(64);
+      // Just under the ceiling is preserved.
+      expect(
+        resolveConcurrencyLimit({ QWEN_CODE_MAX_WORKFLOW_CONCURRENCY: '63' }),
+      ).toBe(63);
+    });
+
+    it('QWEN_CODE_MAX_WORKFLOW_AGENTS actually lowers the cap at run time', async () => {
+      const prev = process.env['QWEN_CODE_MAX_WORKFLOW_AGENTS'];
+      process.env['QWEN_CODE_MAX_WORKFLOW_AGENTS'] = '3';
+      try {
+        const orchestrator = new WorkflowOrchestrator(async () => 'ok');
+        const outcome = await orchestrator.run({
+          script: `return await parallel(
+            Array.from({ length: 4 }, () => () => agent("x"))
+          );`,
+          args: undefined,
+        });
+        const arr = outcome.result as Array<string | null>;
+        expect(arr.filter((v) => v === 'ok')).toHaveLength(3);
+        expect(arr.filter((v) => v === null)).toHaveLength(1);
+      } finally {
+        if (prev === undefined)
+          delete process.env['QWEN_CODE_MAX_WORKFLOW_AGENTS'];
+        else process.env['QWEN_CODE_MAX_WORKFLOW_AGENTS'] = prev;
+      }
+    });
+  });
+});
diff --git a/packages/core/src/agents/runtime/workflow-orchestrator.ts b/packages/core/src/agents/runtime/workflow-orchestrator.ts
index 5a3fcdfe458..e97ee83b322 100644
--- a/packages/core/src/agents/runtime/workflow-orchestrator.ts
+++ b/packages/core/src/agents/runtime/workflow-orchestrator.ts
@@ -5,8 +5,9 @@
  */
 
 import { randomBytes } from 'node:crypto';
+import * as os from 'node:os';
 import type { Config } from '../../config/config.js';
-import { createWorkflowSandbox } from './workflow-sandbox.js';
+import { createWorkflowSandbox, debugLogger } from './workflow-sandbox.js';
 import type {
   WorkflowAgentOpts,
   WorkflowAgentResult,
@@ -14,6 +15,99 @@ import type {
 import { WORKFLOW_SUBAGENT_SYSTEM_PROMPT } from './workflow-prompts.js';
 import { AgentTerminateMode } from './agent-types.js';
 import { ToolNames } from '../../tools/tool-names.js';
+import { createConcurrencyLimiter } from '../../utils/concurrencyLimiter.js';
+
+/**
+ * Default ceiling on total `agent()` calls per workflow run (matches upstream
+ * `hOK = 1000`). Counts EVERY dispatch — sequential, `parallel()`, and
+ * `pipeline()` all funnel through the one wrapped dispatch — so a fan-out
+ * cannot bypass it. The 1001st call throws. Override via env (see below).
+ */
+export const DEFAULT_MAX_AGENTS_PER_RUN = 1000;
+export const MAX_WORKFLOW_AGENTS_ENV = 'QWEN_CODE_MAX_WORKFLOW_AGENTS';
+/**
+ * Absolute upper bound on the env-override agent cap. Even an operator who
+ * sets `QWEN_CODE_MAX_WORKFLOW_AGENTS=999999999` cannot exceed this — the
+ * intent is to catch fat-finger / misconfig that would silently uncap a
+ * runaway workflow (1000-agent default × per-agent token cost). 10000 is
+ * 10× the default, generous for legitimate large fan-outs.
+ */
+export const HARD_MAX_AGENTS_PER_RUN_CEILING = 10_000;
+
+/**
+ * Resolve the per-run agent cap, honoring `QWEN_CODE_MAX_WORKFLOW_AGENTS`.
+ * Mirrors `resolveMaxConcurrentBackgroundAgents` (background-tasks.ts): a
+ * non-integer / <1 override is rejected with a debug warning and the default
+ * is used. An override above `HARD_MAX_AGENTS_PER_RUN_CEILING` is clamped
+ * (with a debug warning) — the env knob is operator-facing, not security-
+ * critical, but a misconfigured ceiling shouldn't silently uncap the run.
+ */
+export function resolveMaxAgentsPerRun(
+  env: Record<string, string | undefined> = process.env,
+): number {
+  const raw = env[MAX_WORKFLOW_AGENTS_ENV];
+  if (raw === undefined || raw.trim() === '') {
+    return DEFAULT_MAX_AGENTS_PER_RUN;
+  }
+  const parsed = Number(raw);
+  if (!Number.isInteger(parsed) || parsed < 1) {
+    debugLogger.warn(
+      `Invalid ${MAX_WORKFLOW_AGENTS_ENV}=${JSON.stringify(raw)}, ` +
+        `using default (${DEFAULT_MAX_AGENTS_PER_RUN})`,
+    );
+    return DEFAULT_MAX_AGENTS_PER_RUN;
+  }
+  if (parsed > HARD_MAX_AGENTS_PER_RUN_CEILING) {
+    debugLogger.warn(
+      `${MAX_WORKFLOW_AGENTS_ENV}=${parsed} exceeds hard ceiling ` +
+        `(${HARD_MAX_AGENTS_PER_RUN_CEILING}); clamping.`,
+    );
+    return HARD_MAX_AGENTS_PER_RUN_CEILING;
+  }
+  return parsed;
+}
+
+export const MAX_WORKFLOW_CONCURRENCY_ENV =
+  'QWEN_CODE_MAX_WORKFLOW_CONCURRENCY';
+/**
+ * Absolute upper bound on the env-override concurrency window. Above this,
+ * a single Node process running N concurrent LLM calls is past the point a
+ * distributed worker is the better tool. 64 ≈ 4× the 16-default ceiling.
+ */
+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.
+ */
+export function resolveConcurrencyLimit(
+  env: Record<string, string | undefined> = process.env,
+): number {
+  const raw = env[MAX_WORKFLOW_CONCURRENCY_ENV];
+  if (raw !== undefined && raw.trim() !== '') {
+    const parsed = Number(raw);
+    if (Number.isInteger(parsed) && parsed >= 1) {
+      if (parsed > HARD_MAX_CONCURRENCY_CEILING) {
+        debugLogger.warn(
+          `${MAX_WORKFLOW_CONCURRENCY_ENV}=${parsed} exceeds hard ceiling ` +
+            `(${HARD_MAX_CONCURRENCY_CEILING}); clamping.`,
+        );
+        return HARD_MAX_CONCURRENCY_CEILING;
+      }
+      return parsed;
+    }
+    debugLogger.warn(
+      `Invalid ${MAX_WORKFLOW_CONCURRENCY_ENV}=${JSON.stringify(raw)}, ` +
+        `using cpu-derived default`,
+    );
+  }
+  return Math.max(1, Math.min(16, os.cpus().length - 2));
+}
 
 /**
  * Bound the resource ceiling for workflow subagents so a single `agent()`
@@ -160,15 +254,53 @@ export class WorkflowOrchestrator {
   constructor(private readonly dispatch: WorkflowAgentDispatch) {}
 
   async run(req: WorkflowRunRequest): Promise<WorkflowRunOutcome> {
-    // Signal threading lives in createProductionDispatch (closure-captured)
-    // rather than per-run state. Sandbox-level signal is intentionally not
-    // exposed in P1 — sync-loop protection is provided by the 30s vm
-    // timeout in workflow-sandbox.ts; async-loop cancellation flows
-    // through dispatch's subagent.execute path.
+    // Signal threading: createProductionDispatch closure-captures a signal
+    // for subagent.execute cancellation. P2 additionally derives a per-run
+    // limiter from req.abortOnTimeout?.signal so wall-clock abort drains
+    // queued dispatches promptly. Sync-loop protection is the 30s vm
+    // timeout in workflow-sandbox.ts; async-loop cancellation flows through
+    // dispatch's subagent.execute path.
     const runId = generateRunId();
+
+    const maxAgents = resolveMaxAgentsPerRun();
+    const signal = req.abortOnTimeout?.signal;
+
+    // P2: the concurrency window throttles AGENT DISPATCHES, not orchestration
+    // thunks. parallel()/pipeline() compose promises freely; only the leaf
+    // agent() calls acquire a slot. This gives the correct "N agents in flight
+    // per run" semantics AND avoids a re-entrancy deadlock (F1, P2 review): if
+    // the window sat at the thunk level, a nested parallel()/pipeline() — e.g.
+    // a pipeline stage that fans out, the canonical /deep-research shape —
+    // would hold every slot while awaiting inner work that can never acquire
+    // one. One shared limiter per run keeps total in-flight agents under the
+    // cap across all fan-out calls.
+    const limiter = createConcurrencyLimiter(resolveConcurrencyLimit(), signal);
+
+    // Every agent() call — sequential, parallel(), or pipeline() — funnels
+    // through this one wrapped dispatch: the counter enforces the per-run agent
+    // cap regardless of launch path (increment-then-check: calls 1..max pass,
+    // the (max+1)th throws), and limiter.run enforces the concurrency window.
+    let agentCount = 0;
+    const countedDispatch: WorkflowAgentDispatch = (prompt, opts) => {
+      agentCount += 1;
+      if (agentCount > maxAgents) {
+        return Promise.reject(
+          new Error(
+            `Workflow exceeded the maximum of ${maxAgents} agent() calls per run.`,
+          ),
+        );
+      }
+      return limiter.run(() => this.dispatch(prompt, opts));
+    };
+
+    const parallelImpl = makeParallelImpl(signal);
+    const pipelineImpl = makePipelineImpl(signal);
+
     const sandbox = createWorkflowSandbox({
       args: req.args,
-      dispatch: this.dispatch,
+      dispatch: countedDispatch,
+      parallel: parallelImpl,
+      pipeline: pipelineImpl,
       abortOnTimeout: req.abortOnTimeout,
     });
     try {
@@ -197,6 +329,162 @@ export class WorkflowOrchestrator {
   }
 }
 
+/**
+ * Settle a batch of thunks into a position-aligned `Array<T|null>` —
+ * errors-as-data: a thunk that rejects (including an over-cap dispatch or a
+ * stage error) becomes `null` at its index, never collapsing the batch.
+ * `Promise.resolve().then(t)` funnels a synchronously-throwing thunk into the
+ * rejection path. The ONE thing that rejects the whole batch is an abort, so
+ * an aborted run surfaces a rejection rather than a silent array of nulls.
+ * Concurrency is bounded at the dispatch layer (limiter.run in countedDispatch),
+ * not here — so nesting a parallel()/pipeline() inside a thunk cannot deadlock.
+ *
+ * Abort responsiveness: this function awaits `Promise.allSettled` which only
+ * settles after every thunk settles. That is NOT a long wait in practice
+ * because the dispatch signal (workflow-orchestrator.ts countedDispatch +
+ * createProductionDispatch) is threaded through to `subagent.execute(ctx,
+ * signal)`, so each in-flight thunk reacts to abort and rejects promptly. The
+ * limiter's separate `addEventListener('abort')` listener drains the
+ * not-yet-started queued thunks instantly. So the apparent "wait for all to
+ * complete" is in reality "wait for all to reach an abort-aware rejection",
+ * which fires immediately after the signal — not after each subagent's full
+ * 10-min internal timeout.
+ */
+async function settleToNullArray(
+  thunks: Array<() => Promise<unknown>>,
+  signal?: AbortSignal,
+): Promise<unknown[]> {
+  const settled = await Promise.allSettled(
+    thunks.map((t) => Promise.resolve().then(t)),
+  );
+  // Use DOMException('AbortError') for consistency with the limiter's
+  // abort path so HOST-side callers seeing this rejection directly can
+  // classify it via isAbortError() (utils/errors.ts). NOTE: this name is
+  // NOT preserved across the vm boundary — vmAsync (workflow-sandbox.ts)
+  // re-throws the script-visible rejection as a fresh vm-realm `new
+  // Error(msg)`, and the orchestrator's outer catch then wraps it as
+  // WorkflowExecutionError. So isAbortError() at the WorkflowTool layer
+  // returns false either way; the DOMException is purely a host-internal
+  // consistency choice, not a script-observable one.
+  if (signal?.aborted)
+    throw new DOMException('Workflow run aborted.', 'AbortError');
+  // Errors-as-data: a rejected thunk becomes null at its index. Log the
+  // discarded rejection reason at debug level so operators investigating a
+  // workflow that returned unexpected nulls can disambiguate between (a) a
+  // dispatch failure (rate limit / model outage), (b) the 1000-agent cap,
+  // (c) a pipeline stage exception, and (d) a non-JSON-serializable thunk
+  // return — all of which surface as the same `null` to the script by
+  // design. The log line is the only operator-side signal of which path
+  // fired; the contract to the script stays opaque.
+  return settled.map((r, i) => {
+    if (r.status === 'fulfilled') return r.value;
+    debugLogger.warn(
+      `Workflow thunk at index ${i} rejected: ${String((r.reason as { message?: unknown })?.message ?? r.reason)}`,
+    );
+    return null;
+  });
+}
+
+/**
+ * Build the host-side `parallel(thunks)` impl. Each thunk is a vm-realm
+ * function whose agent() calls throttle through the per-run concurrency window
+ * at the dispatch layer. A thunk that rejects, or resolves to a non-JSON-
+ * serializable value, becomes `null` at its index (errors-as-data). `parallel()`
+ * itself rejects only when given invalid arguments (non-array / non-function
+ * element) or when the run is aborted. The result array is revived into the
+ * vm realm by the sandbox wrapper (per-element JSON round-trip) — this host
+ * array never reaches the script directly.
+ */
+function makeParallelImpl(
+  signal?: AbortSignal,
+): (thunks: Array<() => Promise<unknown>>) => Promise<unknown[]> {
+  return (thunks) => {
+    if (!Array.isArray(thunks)) {
+      return Promise.reject(
+        new Error(
+          'parallel() expects an array of thunks (functions returning promises).',
+        ),
+      );
+    }
+    for (const t of thunks) {
+      if (typeof t !== 'function') {
+        return Promise.reject(
+          new Error(
+            'parallel() expects an array of functions, not values — wrap each ' +
+              'call: parallel([() => agent(...), () => agent(...)]).',
+          ),
+        );
+      }
+    }
+    return settleToNullArray(thunks, signal);
+  };
+}
+
+/**
+ * Build the host-side `pipeline(items, ...stages)` impl as parallel-of-chains.
+ *
+ * Each item becomes one chain that runs the stages in sequence — staggered,
+ * with NO barrier between stages, so item A can be in stage 3 while item B is
+ * still in stage 1. Stage callbacks receive `(prev, item, idx)`; the first
+ * stage's `prev` is the item itself. A stage that throws, returns `null`, or
+ * returns a non-JSON-serializable value drops that item to `null` and skips
+ * its remaining stages, leaving other items unaffected. Concurrency is
+ * bounded at the dispatch layer, and the result array shares parallel()'s
+ * per-element vm-realm revival.
+ */
+function makePipelineImpl(
+  signal?: AbortSignal,
+): (
+  items: unknown[],
+  ...stages: Array<
+    (prev: unknown, item: unknown, idx: number) => Promise<unknown>
+  >
+) => Promise<unknown[]> {
+  return (items, ...stages) => {
+    if (!Array.isArray(items)) {
+      return Promise.reject(
+        new Error(
+          'pipeline() expects an array of items as its first argument.',
+        ),
+      );
+    }
+    for (const s of stages) {
+      if (typeof s !== 'function') {
+        return Promise.reject(
+          new Error(
+            'pipeline() stages must be functions: ' +
+              'pipeline(items, item => ..., result => ...).',
+          ),
+        );
+      }
+    }
+    const chains = items.map(
+      (item, idx) => () => runPipelineChain(item, idx, stages),
+    );
+    return settleToNullArray(chains, signal);
+  };
+}
+
+/**
+ * Run one item through every stage in order. `null` is the universal drop
+ * sentinel: a stage that returns `null` (or throws — surfaced as a rejection
+ * that the batch maps to `null`) short-circuits the rest of the chain.
+ */
+async function runPipelineChain(
+  item: unknown,
+  idx: number,
+  stages: Array<
+    (prev: unknown, item: unknown, idx: number) => Promise<unknown>
+  >,
+): Promise<unknown> {
+  let prev: unknown = item;
+  for (const stage of stages) {
+    if (prev === null) break;
+    prev = await stage(prev, item, idx);
+  }
+  return prev;
+}
+
 /**
  * Duck-typed message extraction. `instanceof Error` is realm-local; vm-realm
  * Errors raised inside the sandbox are NOT instances of host Error from the
diff --git a/packages/core/src/agents/runtime/workflow-sandbox.test.ts b/packages/core/src/agents/runtime/workflow-sandbox.test.ts
index b6a59e06a6d..dd83ef66fa3 100644
--- a/packages/core/src/agents/runtime/workflow-sandbox.test.ts
+++ b/packages/core/src/agents/runtime/workflow-sandbox.test.ts
@@ -709,28 +709,30 @@ describe('createWorkflowSandbox security', () => {
     expect(['undefined', 'threw', 'no-iterator']).toContain(result);
   });
 
-  // FIX-F (Round 4 UP Critical): P2/P5 primitives are stub-throwing globals
-  // so model-authored scripts get a clear "scheduled for PN" error rather
-  // than `ReferenceError: parallel is not defined` (which the model would
-  // misdiagnose as a bug in its own script).
-  it('parallel() throws a P1-unsupported error rather than ReferenceError', async () => {
+  // FIX-F (Round 4 UP Critical): an un-injected sandbox exposes stub-throwing
+  // parallel/pipeline globals so a model-authored script gets a clear
+  // "unavailable" error rather than `ReferenceError: parallel is not defined`
+  // (which the model would misdiagnose as a bug in its own script). In
+  // production the orchestrator always injects real impls; these stubs only
+  // fire for a bare sandbox constructed without them.
+  it('parallel() throws an availability error rather than ReferenceError when not injected', async () => {
     const sandbox = createWorkflowSandbox({
       args: undefined,
       dispatch: async () => 'ignored',
     });
     await expect(
       sandbox.run(`return parallel([() => agent("a")]);`),
-    ).rejects.toThrow(/parallel.*not supported in P1/);
+    ).rejects.toThrow(/parallel\(\) is unavailable/);
   });
 
-  it('pipeline() throws a P1-unsupported error rather than ReferenceError', async () => {
+  it('pipeline() throws an availability error rather than ReferenceError when not injected', async () => {
     const sandbox = createWorkflowSandbox({
       args: undefined,
       dispatch: async () => 'ignored',
     });
     await expect(
       sandbox.run(`return pipeline([1, 2], x => x, x => x);`),
-    ).rejects.toThrow(/pipeline.*not supported in P1/);
+    ).rejects.toThrow(/pipeline\(\) is unavailable/);
   });
 
   it('workflow() throws a P1-unsupported error rather than ReferenceError', async () => {
@@ -793,6 +795,75 @@ describe('createWorkflowSandbox security', () => {
     expect(result).toEqual([10, 20, 30]);
   });
 
+  // SECURITY (PR #4732 P2): the host parallel/pipeline impl resolves with a
+  // HOST-realm array. vmAsync's resolve path is verbatim, so without the
+  // in-realm JSON revival the RESOLVED array's prototype chain reaches the
+  // host Function constructor — `out.constructor.constructor('return process')()`
+  // would leak host process. The pre-P2 escape test only probed the *Promise*
+  // (vm-realm via vmAsync), NOT the resolved array — this is the uncovered gap.
+  // These tests FAIL against a verbatim wrapper and PASS with revival.
+  it('parallel() RESOLVED array cannot reach host process (revived in-realm)', async () => {
+    const sandbox = createWorkflowSandbox({
+      args: undefined,
+      dispatch: async () => 'ok',
+      // host impl returns a plain HOST array on purpose.
+      parallel: async (thunks) => Promise.all(thunks.map((t) => t())),
+    });
+    const result = await sandbox.run(`
+      const out = await parallel([async () => 1, async () => 2]);
+      try {
+        const v = out.constructor.constructor("return typeof process")();
+        return String(v);
+      } catch (e) { return 'threw:' + String(e.message).slice(0, 40); }
+    `);
+    expect(result).not.toMatch(/object|darwin|linux|win32/i);
+    expect(String(result)).toMatch(/^undefined|^threw/);
+  });
+
+  it('parallel() revives NESTED objects in-realm (not just the outer array)', async () => {
+    const sandbox = createWorkflowSandbox({
+      args: undefined,
+      dispatch: async () => 'ok',
+      parallel: async (thunks) => Promise.all(thunks.map((t) => t())),
+    });
+    const result = await sandbox.run(`
+      const out = await parallel([async () => ({ k: 'v' })]);
+      try {
+        const v = out[0].constructor.constructor("return typeof process")();
+        return String(v);
+      } catch (e) { return 'threw:' + String(e.message).slice(0, 40); }
+    `);
+    expect(result).not.toMatch(/object|darwin|linux|win32/i);
+    expect(String(result)).toMatch(/^undefined|^threw/);
+  });
+
+  it('pipeline() RESOLVED array cannot reach host process (revived in-realm)', async () => {
+    const sandbox = createWorkflowSandbox({
+      args: undefined,
+      dispatch: async () => 'ok',
+      pipeline: async (items, ...stages) => {
+        const out: unknown[] = [];
+        for (let i = 0; i < items.length; i++) {
+          let cur: unknown = items[i];
+          for (const stage of stages) {
+            cur = await stage(cur, items[i], i);
+          }
+          out.push(cur);
+        }
+        return out;
+      },
+    });
+    const result = await sandbox.run(`
+      const out = await pipeline([1, 2], async (x) => x * 10);
+      try {
+        const v = out.constructor.constructor("return typeof process")();
+        return String(v);
+      } catch (e) { return 'threw:' + String(e.message).slice(0, 40); }
+    `);
+    expect(result).not.toMatch(/object|darwin|linux|win32/i);
+    expect(String(result)).toMatch(/^undefined|^threw/);
+  });
+
   it('opts.budget overrides the throwing stub when provided', async () => {
     const sandbox = createWorkflowSandbox({
       args: undefined,
diff --git a/packages/core/src/agents/runtime/workflow-sandbox.ts b/packages/core/src/agents/runtime/workflow-sandbox.ts
index c2b08851cc9..84a42e3ea01 100644
--- a/packages/core/src/agents/runtime/workflow-sandbox.ts
+++ b/packages/core/src/agents/runtime/workflow-sandbox.ts
@@ -120,6 +120,13 @@ function isRegexContext(source: string, i: number): boolean {
 }
 
 import * as vm from 'node:vm';
+import { createDebugLogger } from '../../utils/debugLogger.js';
+
+// Shared with workflow-orchestrator (avoids a duplicate createDebugLogger
+// instance with the same 'WORKFLOW' namespace). Re-exported so orchestrator
+// imports the same instance — orchestrator already imports from this module,
+// so this is the natural direction (the reverse would be a circular dep).
+export const debugLogger = createDebugLogger('WORKFLOW');
 
 // Cap log + phase lines to prevent unbounded memory growth from runaway
 // model-authored loops.
@@ -220,11 +227,16 @@ export interface SandboxOptions {
 }
 
 /**
- * T23 (PR #4732 R2): default async wall-clock cap. 30 minutes is generous
- * for any realistic P1 sequential workflow (single agent capped at
- * 10 min × ~10 agents max practical → ~100 min upper bound, but typical
- * workflows finish in seconds). 30 min stops 0-token hang patterns
- * before they waste operator hours.
+ * T23 (PR #4732 R2): default async wall-clock cap. The wall clock is a
+ * 0-token-hang backstop, NOT a precise cost cap: it bounds patterns like an
+ * in-script `await new Promise(() => {})` that the vm timeout cannot reach.
+ * For genuine cost control, use the env-overridable per-run cap
+ * (`QWEN_CODE_MAX_WORKFLOW_AGENTS`) and concurrency window
+ * (`QWEN_CODE_MAX_WORKFLOW_CONCURRENCY`). 30 minutes is set generously
+ * enough that typical workflows never see it but a hang doesn't waste
+ * operator hours; raise via `QWEN_CODE_MAX_WORKFLOW_SECONDS` for long
+ * legitimate fan-outs (1000 agents × 10-min subagent cap ÷ default
+ * concurrency would already exceed 30 min).
  */
 const DEFAULT_MAX_WALL_CLOCK_MS = 30 * 60 * 1000;
 
@@ -341,6 +353,18 @@ export function createWorkflowSandbox(opts: SandboxOptions): WorkflowSandbox {
     pushLog: safeLog,
     lastPhase: () => phases[phases.length - 1],
     hostAgent: opts.dispatch,
+    // PR #4947 R2 T7 (qwen-code-ci-bot): host-side log hook for reviveInRealm's
+    // catch path. Mirrors the rejection-logging in settleToNullArray so an
+    // operator running with debug logging can distinguish "thunk rejected"
+    // (settleToNullArray.warn) from "thunk resolved to a non-JSON-serializable
+    // value" (this warn). Receives only primitive strings/numbers — the bridge
+    // contract forbids host objects crossing back to the script.
+    logRevivalFailure: (idx: number, reason: string): void => {
+      debugLogger.warn(
+        `Workflow result revival failed at index ${idx}: ${reason}; ` +
+          `slot set to null (non-JSON-serializable thunk return).`,
+      );
+    },
     // The truthy flags distinguish "injected" from "default stub" inside the
     // init script without leaking the host function itself when not used.
     hasParallel: !!opts.parallel,
@@ -510,32 +534,80 @@ export function createWorkflowSandbox(opts: SandboxOptions): WorkflowSandbox {
       });
 
       // --- parallel / pipeline ---
+      // SECURITY (PR #4732 P2): the host impl resolves with a HOST-realm array.
+      // vmAsync's resolve path is verbatim (it does NOT re-wrap resolved
+      // values), so handing that host array to the script would reopen the
+      // T1/T8/T14 escape: result.constructor.constructor('return process')()
+      // walks the host Array.prototype chain to the host Function constructor.
+      // We revive the array INSIDE the vm realm with JSON.parse(JSON.stringify)
+      // -- the same mechanism that makes the args global safe (see the args
+      // revival above) -- so the value the script sees has vm-realm prototypes
+      // whose constructors can't reach host process. Agent results are JSON
+      // strings (and null slots), so the round-trip is lossless for P2.
+      //
+      // EAD-1 (P2 self-review): revive PER-ELEMENT, not the whole array in one
+      // JSON.stringify. A single slot whose VALUE is non-serializable (a thunk
+      // that returns a BigInt or a circular object) must become null at its
+      // index -- it must NOT throw on the whole array and destroy every sibling
+      // result, which would defeat errors-as-data for return values. The outer
+      // [] is built in-realm here, so the result keeps vm-realm prototypes.
+      //
+      // SECURITY (PR #4947 R1 wenshao): reviveInRealm MUST remain inside this
+      // vm init runInContext block. JSON, Array, Object here are vm-realm
+      // globals; extracting this function to a host-side utility (e.g. a
+      // shared utils/jsonRevive.ts) would resolve those references against
+      // the HOST realm, silently reopening the T1/T8/T14 escape that the
+      // revival is designed to prevent. The textual identity to a host-side
+      // util is exactly the trap.
+      function reviveInRealm(hostArr) {
+        const out = [];
+        for (let i = 0; i < hostArr.length; i++) {
+          try {
+            out[i] = JSON.parse(JSON.stringify(hostArr[i]));
+          } catch (e) {
+            // Cross to host realm for debug logging. The bridge function
+            // accepts only primitive strings/numbers; the error message is
+            // coerced to a String here so no vm-realm Error object crosses.
+            __b.logRevivalFailure(i, String(e?.message ?? e));
+            out[i] = null;
+          }
+        }
+        return out;
+      }
       if (__b.hasParallel) {
-        globalThis.parallel = vmAsync(function (thunks) {
+        const callParallel = vmAsync(function (thunks) {
           return __b.hostParallel(thunks);
         });
+        globalThis.parallel = function parallel(thunks) {
+          return callParallel(thunks).then(reviveInRealm);
+        };
       } else {
         globalThis.parallel = function parallel() {
           return new Promise(function (_, reject) {
             reject(new Error(
-              'parallel() is not supported in P1. Sequential agent() is the only ' +
-              'execution mode in P1. Concurrent fan-out is scheduled for P2.'
+              'parallel() is unavailable: this sandbox was created without a ' +
+              'parallel implementation. The orchestrator injects one; a bare ' +
+              'sandbox has no concurrent-dispatch capability.'
             ));
           });
         };
       }
       if (__b.hasPipeline) {
-        globalThis.pipeline = vmAsync(function (items) {
+        const callPipeline = vmAsync(function (items) {
           const stages = [];
           for (let i = 1; i < arguments.length; i++) stages.push(arguments[i]);
           return __b.hostPipeline.apply(null, [items].concat(stages));
         });
+        globalThis.pipeline = function pipeline() {
+          return callPipeline.apply(null, arguments).then(reviveInRealm);
+        };
       } else {
         globalThis.pipeline = function pipeline() {
           return new Promise(function (_, reject) {
             reject(new Error(
-              'pipeline() is not supported in P1. Staggered multi-stage execution ' +
-              'is scheduled for P2.'
+              'pipeline() is unavailable: this sandbox was created without a ' +
+              'pipeline implementation. The orchestrator injects one; a bare ' +
+              'sandbox has no staggered multi-stage capability.'
             ));
           });
         };
diff --git a/packages/core/src/tools/workflow/workflow.test.ts b/packages/core/src/tools/workflow/workflow.test.ts
index 30667dc1526..f6302e743b9 100644
--- a/packages/core/src/tools/workflow/workflow.test.ts
+++ b/packages/core/src/tools/workflow/workflow.test.ts
@@ -64,6 +64,41 @@ describe('WorkflowTool', () => {
     expect(JSON.stringify(result.returnDisplay)).toMatch(/wf_[0-9a-f]{16}/);
   });
 
+  // P2 (PR #4732): parallel() runs end-to-end through the full stack
+  // (WorkflowTool → orchestrator counter+limiter+parallelImpl → sandbox
+  // in-realm revival → script return → safeStringifyResult).
+  it('execute() runs parallel() end-to-end and returns the revived array', async () => {
+    const tool = new WorkflowTool(fakeConfig(), {
+      dispatch: async (prompt) => `T:${prompt}`,
+    });
+    const invocation = tool.build({
+      script: `return await parallel([() => agent("a"), () => agent("b")]);`,
+    });
+    const result = await invocation.execute(new AbortController().signal);
+    expect(result.error).toBeUndefined();
+    const llmText = (result.llmContent as Array<{ text: string }>)[0].text;
+    expect(JSON.parse(llmText)).toEqual(['T:a', 'T:b']);
+  });
+
+  // PR #4947 R2 T8 (qwen-code-ci-bot): pipeline() through WorkflowTool
+  // exercises a vm wrapper path that is structurally distinct from parallel's
+  // single-argument call — pipeline uses `callPipeline.apply(null, arguments)`
+  // and `[items].concat(stages)` to spread the variadic stage list
+  // (workflow-sandbox.ts pipeline wrapper). A regression in the vm-to-host
+  // stage forwarding would not be caught by the parallel E2E test above.
+  it('execute() runs pipeline() end-to-end and returns the revived array', async () => {
+    const tool = new WorkflowTool(fakeConfig(), {
+      dispatch: async () => 'unused',
+    });
+    const invocation = tool.build({
+      script: `return await pipeline([1, 2], (x) => x * 10, (x) => x + 1);`,
+    });
+    const result = await invocation.execute(new AbortController().signal);
+    expect(result.error).toBeUndefined();
+    const llmText = (result.llmContent as Array<{ text: string }>)[0].text;
+    expect(JSON.parse(llmText)).toEqual([11, 21]);
+  });
+
   // TST-C3: execute() should return an error result (not throw) when the script throws.
   it('execute() returns an error result when the script throws', async () => {
     const tool = new WorkflowTool(fakeConfig(), {
diff --git a/packages/core/src/tools/workflow/workflow.ts b/packages/core/src/tools/workflow/workflow.ts
index a98b4650409..7745cbf6b94 100644
--- a/packages/core/src/tools/workflow/workflow.ts
+++ b/packages/core/src/tools/workflow/workflow.ts
@@ -6,7 +6,8 @@
 
 /**
  * @fileoverview WorkflowTool — user-facing tool that executes a workflow script
- * via WorkflowOrchestrator (P1: sequential agent dispatch only).
+ * via WorkflowOrchestrator. Supports sequential `agent()`, plus concurrent
+ * fan-out via `parallel()` / `pipeline()` throttled at the dispatch layer.
  */
 
 import {
@@ -56,10 +57,20 @@ const WORKFLOW_PARAM_SCHEMA = {
         'JavaScript source of the workflow. Wrapped as an async IIFE. ' +
         'May call the injected globals `phase(title)`, `log(msg)`, ' +
         '`agent(prompt, { label? })`, and read `args`. ' +
-        'P1 ships sequential primitives only — `parallel()` and `pipeline()` ' +
-        'arrive in P2. Writing `Promise.all([agent(), agent()])` spawns ' +
-        'concurrent subagents that share Config and may race on file edits; ' +
-        'prefer `await agent(...)` for ordered execution. ' +
+        'Concurrency: `parallel([() => agent(...), ...])` runs thunks ' +
+        'through a shared per-run window (default ' +
+        '`max(1, min(16, cpus-2))` agents in flight; override via ' +
+        '`QWEN_CODE_MAX_WORKFLOW_CONCURRENCY`) and resolves to a ' +
+        'position-aligned array — a thunk that throws, or resolves to a ' +
+        'non-JSON-serializable value, becomes `null` at its index ' +
+        '(errors-as-data); parallel() itself rejects only on invalid ' +
+        'arguments or abort. `pipeline(items, ...stages)` runs each item ' +
+        'through the stages (staggered, no inter-stage barrier); a stage ' +
+        'that throws, returns `null`, or returns a non-JSON-serializable ' +
+        'value drops that item to `null`. Pass ' +
+        'THUNKS to parallel, not eager calls: `parallel([() => agent(...)])`, ' +
+        'not `parallel([agent(...)])`. At most 1000 agent() calls per run ' +
+        '(override via `QWEN_CODE_MAX_WORKFLOW_AGENTS`). ' +
         '`Date.now()` and `Math.random()` both throw — workflow scripts ' +
         'must be deterministic for resume. ' +
         '`export const meta = {...}` declarations are stripped before execution.',
@@ -250,11 +261,14 @@ export class WorkflowTool extends BaseDeclarativeTool<
     super(
       ToolNames.WORKFLOW,
       ToolDisplayNames.WORKFLOW,
-      'Execute a workflow script that orchestrates subagents sequentially. ' +
-        'P1 supports `phase`, `log`, and sequential `agent` only. No parallel, ' +
-        'no pipeline, no schema, no resume, no background execution. ' +
-        'Scripts run in a node:vm sandbox without access to the filesystem or ' +
-        'shell; all I/O happens through the spawned agents.',
+      'Execute a workflow script that orchestrates subagents. ' +
+        'Supports `phase`, `log`, sequential `agent`, and concurrent fan-out ' +
+        'via `parallel(thunks)` / `pipeline(items, ...stages)` (default ' +
+        '`max(1, min(16, cpus-2))` agents in flight per run, up to 1000 ' +
+        'agents total; both env-overridable). No schema, no resume, no ' +
+        'background execution yet. Scripts run in a node:vm sandbox without ' +
+        'access to the filesystem or shell; all I/O happens through the ' +
+        'spawned agents.',
       Kind.Other,
       WORKFLOW_PARAM_SCHEMA,
       /* isOutputMarkdown */ true,
diff --git a/packages/core/src/utils/concurrencyLimiter.test.ts b/packages/core/src/utils/concurrencyLimiter.test.ts
new file mode 100644
index 00000000000..89c405c2507
--- /dev/null
+++ b/packages/core/src/utils/concurrencyLimiter.test.ts
@@ -0,0 +1,117 @@
+/**
+ * @license
+ * Copyright 2025 Qwen
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import { describe, it, expect } from 'vitest';
+import { createConcurrencyLimiter } from './concurrencyLimiter.js';
+
+describe('createConcurrencyLimiter', () => {
+  it('throws on a non-positive-integer limit', () => {
+    expect(() => createConcurrencyLimiter(0)).toThrow(/positive integer/i);
+    expect(() => createConcurrencyLimiter(-1)).toThrow(/positive integer/i);
+    expect(() => createConcurrencyLimiter(1.5)).toThrow(/positive integer/i);
+    expect(() => createConcurrencyLimiter(Number.NaN)).toThrow(
+      /positive integer/i,
+    );
+  });
+
+  describe('run', () => {
+    it('resolves with the thunk value', async () => {
+      const limiter = createConcurrencyLimiter(2);
+      await expect(limiter.run(async () => 42)).resolves.toBe(42);
+    });
+
+    it('propagates a thunk rejection raw (no null coercion)', async () => {
+      const limiter = createConcurrencyLimiter(2);
+      await expect(
+        limiter.run(async () => {
+          throw new Error('boom');
+        }),
+      ).rejects.toThrow('boom');
+    });
+
+    it('never runs more than `limit` thunks concurrently', async () => {
+      const limit = 3;
+      const limiter = createConcurrencyLimiter(limit);
+      let active = 0;
+      let peak = 0;
+      const mk = () => async () => {
+        active++;
+        peak = Math.max(peak, active);
+        await new Promise((r) => setTimeout(r, 5));
+        active--;
+        return 'x';
+      };
+      await Promise.all(
+        Array.from({ length: 12 }, mk).map((t) => limiter.run(t)),
+      );
+      // 12 thunks, window 3, all want to run → steady-state fills the window.
+      expect(peak).toBe(limit);
+    });
+
+    it('processes the full queue (every thunk eventually runs)', async () => {
+      const limiter = createConcurrencyLimiter(2);
+      let ran = 0;
+      await Promise.all(
+        Array.from({ length: 20 }, () => () => {
+          ran++;
+          return Promise.resolve('ok');
+        }).map((t) => limiter.run(t)),
+      );
+      expect(ran).toBe(20);
+    });
+  });
+
+  describe('abort', () => {
+    it('run() rejects immediately when the signal is already aborted', async () => {
+      const ac = new AbortController();
+      ac.abort();
+      const limiter = createConcurrencyLimiter(2, ac.signal);
+      await expect(limiter.run(async () => 1)).rejects.toThrow(/abort/i);
+    });
+
+    // Defensive: queued jobs must be rejected the moment the signal fires,
+    // NOT lazily on the next in-flight settlement. Otherwise a non-settling
+    // in-flight thunk (a buggy/hung future dispatcher) would wedge every
+    // queued job forever. Production today wouldn't hit this because
+    // subagent.execute always settles, but the limiter shouldn't lean on
+    // an unenforced invariant.
+    it('rejects queued jobs promptly when the signal aborts, even if in-flight never settles', async () => {
+      const ac = new AbortController();
+      const limiter = createConcurrencyLimiter(1, ac.signal);
+      // Hold the one slot with a thunk that never settles.
+      limiter.run(() => new Promise<never>(() => {})).catch(() => {});
+      // Queue a second job; with abort it must reject promptly, not wait for
+      // the hung in-flight to settle.
+      const queued = limiter.run(async () => 'never');
+      setTimeout(() => ac.abort(), 5);
+      await expect(
+        Promise.race([
+          queued,
+          new Promise((_, rej) =>
+            setTimeout(() => rej(new Error('hung')), 200),
+          ),
+        ]),
+      ).rejects.toThrow(/abort/i);
+    });
+
+    it('does not start queued thunks after abort', async () => {
+      const ac = new AbortController();
+      const limiter = createConcurrencyLimiter(1, ac.signal);
+      let started = 0;
+      const thunks = Array.from({ length: 5 }, () => async () => {
+        started++;
+        await new Promise((r) => setTimeout(r, 10));
+        return 1;
+      });
+      // One slot; first thunk starts, the rest queue behind it.
+      const ps = thunks.map((t) => limiter.run(t).catch(() => {}));
+      setTimeout(() => ac.abort(), 5);
+      await Promise.all(ps);
+      // The first thunk had started; abort must prevent the queued 4 from starting.
+      expect(started).toBeLessThan(5);
+    });
+  });
+});
diff --git a/packages/core/src/utils/concurrencyLimiter.ts b/packages/core/src/utils/concurrencyLimiter.ts
new file mode 100644
index 00000000000..ef2a464940d
--- /dev/null
+++ b/packages/core/src/utils/concurrencyLimiter.ts
@@ -0,0 +1,111 @@
+/**
+ * @license
+ * Copyright 2025 Qwen
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+/**
+ * @fileoverview A shared sliding-window concurrency limiter (p-limit style).
+ *
+ * Unlike the fixed-size batch loops elsewhere in the codebase, this keeps a
+ * window of at most `limit` thunks in flight and starts a queued thunk the
+ * instant a slot frees — so a single instance can be SHARED across many
+ * callers (e.g. every agent() dispatch within one workflow run) and still
+ * hold the total in-flight count under one cap.
+ */
+
+/**
+ * Thrown (and surfaced as a rejection) when the limiter is aborted via its
+ * `AbortSignal`. Named `AbortError` so `isAbortError()` (utils/errors.ts) can
+ * classify it the same way as other cancellations in the codebase.
+ */
+function abortError(): Error {
+  // DOMException is a global in Node ≥17 and is already used elsewhere in this
+  // package for the same purpose (utils/abortController.ts). The 'AbortError'
+  // name is what isAbortError() keys off.
+  return new DOMException('Concurrency limiter aborted.', 'AbortError');
+}
+
+export interface ConcurrencyLimiter {
+  /**
+   * Schedule a single thunk. Resolves/rejects with the thunk's own
+   * settlement (rejections are propagated raw — the caller decides whether
+   * to treat them as data). At most `limit` scheduled thunks run at once,
+   * across ALL callers sharing this limiter.
+   */
+  run<T>(thunk: () => Promise<T>): Promise<T>;
+}
+
+export function createConcurrencyLimiter(
+  limit: number,
+  signal?: AbortSignal,
+): ConcurrencyLimiter {
+  if (!Number.isInteger(limit) || limit < 1) {
+    throw new Error(
+      `Concurrency limit must be a positive integer, got ${String(limit)}.`,
+    );
+  }
+
+  let active = 0;
+  type Job = {
+    thunk: () => Promise<unknown>;
+    resolve: (v: unknown) => void;
+    reject: (e: unknown) => void;
+  };
+  const queue: Job[] = [];
+
+  const pump = (): void => {
+    while (active < limit && queue.length > 0) {
+      const job = queue.shift()!;
+      // Don't start NEW work once aborted — reject the job without invoking
+      // its thunk so an aborted run can't keep spawning agents. In-flight
+      // thunks are cancelled through their own signal (the dispatch layer).
+      if (signal?.aborted) {
+        job.reject(abortError());
+        continue;
+      }
+      active++;
+      // Promise.resolve().then(thunk) so a thunk that throws synchronously is
+      // funnelled into the rejection path rather than escaping pump().
+      Promise.resolve()
+        .then(job.thunk)
+        .then(job.resolve, job.reject)
+        .finally(() => {
+          active--;
+          pump();
+        });
+    }
+  };
+
+  const run = <T>(thunk: () => Promise<T>): Promise<T> =>
+    new Promise<T>((resolve, reject) => {
+      if (signal?.aborted) {
+        reject(abortError());
+        return;
+      }
+      queue.push({
+        thunk: thunk as () => Promise<unknown>,
+        resolve: resolve as (v: unknown) => void,
+        reject,
+      });
+      pump();
+    });
+
+  // Drain the queue the moment the signal fires, rather than waiting for
+  // some in-flight thunk's finally to re-run pump(). Without this, a thunk
+  // that never settles would hold the slot and wedge every queued job
+  // forever. Cleanup matters: a listener on a long-lived parent signal that
+  // outlives many limiters would leak — `{ once: true }` covers it (abort
+  // fires at most once per AbortSignal).
+  if (signal && !signal.aborted) {
+    signal.addEventListener(
+      'abort',
+      () => {
+        while (queue.length > 0) queue.shift()!.reject(abortError());
+      },
+      { once: true },
+    );
+  }
+
+  return { run };
+}