fix: parallel/pipeline 不应吞掉工作流取消异常

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(

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 {

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}`,
           )