From b7d8458a765238bcc924b5151e18c16b6bc91535 Mon Sep 17 00:00:00 2001
From: Rui <1685901819@qq.com>
Date: Tue, 11 Aug 2026 06:19:05 +0800
Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=9D=E7=95=99=E5=B7=A5=E4=BD=9C?=
 =?UTF-8?q?=E6=B5=81=E5=8F=96=E6=B6=88=E7=8A=B6=E6=80=81?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../src/__tests__/hooks.test.ts               | 20 +++++++++++
 .../src/__tests__/runWorkflow.test.ts         | 35 +++++++++++++++++++
 packages/workflow-engine/src/engine/hooks.ts  |  7 ++++
 3 files changed, 62 insertions(+)

diff --git a/packages/workflow-engine/src/__tests__/hooks.test.ts b/packages/workflow-engine/src/__tests__/hooks.test.ts
index 539595bc2..9453fb2b5 100644
--- a/packages/workflow-engine/src/__tests__/hooks.test.ts
+++ b/packages/workflow-engine/src/__tests__/hooks.test.ts
@@ -617,6 +617,17 @@ test('parallel single item throws → logger.warn records the failure reason', a
   expect(warns[0]).toMatch(/boom-x/)
 })
 
+test('parallel rethrows workflow cancellation instead of converting it to null', async () => {
+  const { hooks } = buildCtx()
+  await expect(
+    hooks.parallel([
+      async () => {
+        throw new WorkflowAbortedError()
+      },
+    ]),
+  ).rejects.toBeInstanceOf(WorkflowAbortedError)
+})
+
 test('pipeline chains stage by stage, stage throws → null', async () => {
   const { hooks } = buildCtx()
   const out = await hooks.pipeline(
@@ -645,6 +656,15 @@ test('pipeline stage throws → logger.warn records the failure reason', async (
   expect(warns[0]).toMatch(/stage-boom/)
 })
 
+test('pipeline rethrows workflow cancellation instead of converting it to null', async () => {
+  const { hooks } = buildCtx()
+  await expect(
+    hooks.pipeline([1], async () => {
+      throw new WorkflowAbortedError()
+    }),
+  ).rejects.toBeInstanceOf(WorkflowAbortedError)
+})
+
 test('pipeline over 4096 throws', async () => {
   const { hooks } = buildCtx()
   await expect(
diff --git a/packages/workflow-engine/src/__tests__/runWorkflow.test.ts b/packages/workflow-engine/src/__tests__/runWorkflow.test.ts
index d282bc73d..5312e33d6 100644
--- a/packages/workflow-engine/src/__tests__/runWorkflow.test.ts
+++ b/packages/workflow-engine/src/__tests__/runWorkflow.test.ts
@@ -192,6 +192,41 @@ test('abort → killed', async () => {
   }
 })
 
+test('abort inside parallel and pipeline remains killed through run_done', async () => {
+  const dir = await mkdtemp(join(tmpdir(), 'wf-run-'))
+  try {
+    const cases = [
+      ['parallel', `return parallel([() => agent('x')])`],
+      ['pipeline', `return pipeline(['x'], (_prev, item) => agent(item))`],
+    ] as const
+
+    for (const [name, script] of cases) {
+      const { ports, events } = portsWithEvents(dir, new Map())
+      const ac = new AbortController()
+      ac.abort()
+      const result = await runWorkflow({
+        script,
+        runId: `run-abort-${name}`,
+        ports,
+        host: createHostHandle(null),
+        signal: ac.signal,
+        cwd: dir,
+        budgetTotal: null,
+      })
+
+      expect(result.status).toBe('killed')
+      expect(events.findLast(event => event.type === 'run_done')).toMatchObject(
+        {
+          type: 'run_done',
+          status: 'killed',
+        },
+      )
+    }
+  } finally {
+    await rm(dir, { recursive: true, force: true })
+  }
+})
+
 test('workflow() nesting (one level) shares counts', async () => {
   const dir = await mkdtemp(join(tmpdir(), 'wf-run-'))
   try {
diff --git a/packages/workflow-engine/src/engine/hooks.ts b/packages/workflow-engine/src/engine/hooks.ts
index 0e9670337..f87ee402a 100644
--- a/packages/workflow-engine/src/engine/hooks.ts
+++ b/packages/workflow-engine/src/engine/hooks.ts
@@ -273,6 +273,10 @@ export function makeHooks(
         try {
           return await t()
         } catch (e) {
+          // Cancellation is control flow for the entire run, not an item-level
+          // failure. Swallowing it here makes runWorkflow persist a killed run
+          // as successfully completed with a null item.
+          if (e instanceof WorkflowAbortedError) throw e
           // The "null on error" contract is unchanged, but it should log — otherwise the workflow author cannot locate why an agent failed
           ctx.ports.logger.warn?.(
             `parallel thunk #${i} failed: ${(e as Error).message}`,
@@ -303,6 +307,9 @@ export function makeHooks(
           }
           return prev as R
         } catch (e) {
+          // Keep user cancellation observable by runWorkflow so it can emit
+          // and persist the terminal `killed` state.
+          if (e instanceof WorkflowAbortedError) throw e
           ctx.ports.logger.warn?.(
             `pipeline item #${index} failed: ${(e as Error).message}`,
           )