root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix(engine): honor graph cancellation at the merge node

Browse Source 
The merge node could not observe a graph abort. WorkflowPrimitiveContext
carried no signal, so requestMerge raced the merge only against its own
30-minute GRAPH_MERGE_TIMEOUT_MS using a controller it owned. A hard-cancel
(user cancel, engine restart, pause/resume) aborted the graph controller and
the walk kept sitting inside the merge node for the full timeout. When the
timeout finally fired it aborted the still-running AI merge -- surfacing as
"Manual-merge failed: Request was aborted" -- and the walk reported
value=merge-timeout for a cancellation it had missed half an hour earlier.
An abort landing between merger-ai's `worktree: null` write and
mergeConfirmed then stranded the card as no-worktree-no-merge-confirmed.

Thread the graph AbortSignal from WorkflowNodeExecutionContext (where it
already existed) through primitiveNodeContext/primitiveContextForNode into
the primitives, and honor it on both merge surfaces:

- requestMerge fails fast when the walk is already cancelled, before
  ensureWorkflowMergeBoundaryTask mutates the row or the requester enqueues
  a merge, and links the graph signal into its timeout controller via
  AbortSignal.any -- raced separately so the walk returns on the abort
  rather than waiting on a requester that may never settle.
- The legacy merge seam had the identical unguarded race and gets the same
  treatment.

The timeout stays: it bounds a wedged merge queue, which is a different
failure from cancellation. Both signals must stay live -- dropping either
silently restores the stall with no type error.

Cancellation returns a distinct `merge-cancelled` rather than reusing
merge-timeout. Returning `data.status: "failed"` would let classifyMergeFailure
read the unknown reason as merge-failed and route the cancellation into
bounded auto-merge retry, re-requesting the merge the operator just cancelled.

Regression test covers both merge surfaces, both cancel timings (pre-flight
and mid-flight), the no-signal back-compat path, the signal plumbing itself,
and the classification boundary. Verified by removing the fix: 7 of 9 cases
fail, with the mid-flight cases hanging until timeout.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsxdsm
2026-07-15 19:43:08 -07:00
parent 1043e44bc2
commit 753b1bb710
6 changed files with 317 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
.changeset/merge-node-honors-graph-cancellation.md Normal file
View File

@@ -0,0 +1,7 @@
---
"@runfusion/fusion": patch
---
summary: Cancelling a merging task now stops it immediately instead of stalling for 30 minutes.
category: fix
dev: The `merge` runtime primitive and legacy merge seam raced the merge only against their own 30-minute `GRAPH_MERGE_TIMEOUT_MS`, never observing the graph's abort — `WorkflowPrimitiveContext` had no `signal`. Threads the graph `AbortSignal` through `primitiveNodeContext`/`primitiveContextForNode` into both merge paths (linked via `AbortSignal.any`, timeout preserved as the wedged-queue bound) and returns a distinct `merge-cancelled` value that does not route into bounded auto-merge retry.

244
packages/engine/src/__tests__/workflow-merge-cancellation.test.ts Normal file
View File

@@ -0,0 +1,244 @@
// @ts-nocheck
/*
FNXC:WorkflowCancellation 2026-07-15-10:42:
Regression cover for the graph-cancellation invariant: a cancelled graph walk must collapse an in-flight merge IMMEDIATELY, never sit inside the merge node until its own 30-minute timeout fires.
Original symptom: a hard-cancel aborted the graph controller while the `merge` node was in flight. The merge primitive raced the merge only against `GRAPH_MERGE_TIMEOUT_MS` (30 min) using a controller it owned, so it never observed the cancel. Thirty minutes later the timeout fired, aborted the still-running AI merge ("Manual-merge failed: Request was aborted"), and the walk finally discovered it had been cancelled half an hour earlier — reported as `value=merge-timeout`. An abort landing between merger-ai's `worktree: null` write and `mergeConfirmed` then stranded the card as `no-worktree-no-merge-confirmed`.
Surface enumeration (engine-only; no UI, so desktop/mobile breakpoints are N/A):
- Both merge surfaces: the `requestMerge` runtime primitive AND the legacy merge seam (`createAuthoritativeWorkflowSeams().merge`), which had the identical unguarded 30-minute race.
- Both cancel timings: signal already aborted at entry (pre-flight) and aborted mid-flight.
- Data states: signal present vs absent (absent must preserve pre-fix behavior).
- The plumbing that feeds both: `primitiveNodeContext` / the node-handler context builder / the merge runner — an unthreaded signal silently reintroduces the stall with no type error.
- The classification boundary: `merge-cancelled` must not be read as a retryable merge failure.
*/
import { describe, it, expect, vi, beforeEach } from "vitest";
import "./executor-test-helpers.js";
import { TaskExecutor } from "../executor.js";
import { primitiveNodeContext } from "../runtime-primitives.js";
import { classifyMergePrimitiveResult } from "../workflow-merge-nodes.js";
import { createMergeAttemptHandler } from "../workflow-node-runners/merge-runner.js";
import { createMockStore, mockedExistsSync, resetExecutorMocks } from "./executor-test-helpers.js";
const now = "2026-07-15T00:00:00.000Z";
/** A task shaped to clear the merge boundary's implementation-proof gates, so the
* cancellation race — not a pre-flight rejection — is what the assertion observes. */
function mergeReadyTask(overrides = {}) {
return {
id: "FN-CANCEL",
title: "Cancellable merge task",
description: "exercise graph cancellation at the merge node",
column: "in-progress",
dependencies: [],
steps: [],
currentStep: 0,
log: [],
noCommitsExpected: true,
branch: null,
worktree: null,
enabledWorkflowSteps: [],
prompt: "# Task\n\n## Steps\n\n### Step 1: Decide\n- [ ] Record no-code decision",
createdAt: now,
updatedAt: now,
...overrides,
};
}
function executorFor(liveTask) {
const store = createMockStore();
store.getTask.mockResolvedValue(liveTask);
store.moveTask.mockResolvedValue({ ...liveTask, column: "in-review" });
const executor = new TaskExecutor(store, "/tmp/test");
return { store, executor };
}
function mergeCtx(signal) {
return primitiveNodeContext(
{ runId: "FN-CANCEL:run", taskId: "FN-CANCEL", workflowId: "builtin:coding" },
{ id: "merge", kind: "prompt" },
{},
signal,
);
}
/** A merge requester that never settles — it stands in for an AI merge still in
* flight. If cancellation is not observed, awaiting the primitive hangs and the
* test times out, which is exactly the production stall. */
function pendingMergeRequester() {
const calls = [];
const requester = vi.fn((taskId, options) => {
calls.push({ taskId, signal: options?.signal });
return new Promise(() => {});
});
return { requester, calls };
}
describe("workflow merge cancellation", () => {
beforeEach(() => {
resetExecutorMocks();
mockedExistsSync.mockReturnValue(true);
});
describe("requestMerge primitive", () => {
it("collapses immediately when the graph aborts mid-merge instead of waiting for the 30-minute timeout", async () => {
const liveTask = mergeReadyTask();
const { executor } = executorFor(liveTask);
const { requester, calls } = pendingMergeRequester();
executor.setMergeRequester(requester);
const controller = new AbortController();
const pending = executor
.createAuthoritativeWorkflowPrimitives({ autoMerge: true })
.requestMerge(mergeCtx(controller.signal), liveTask);
// Let the primitive reach the merge requester before cancelling.
await vi.waitFor(() => expect(requester).toHaveBeenCalled());
controller.abort();
// No timer advance: the walk must return on the abort itself.
await expect(pending).resolves.toMatchObject({ outcome: "failure", value: "merge-cancelled" });
// The in-flight merge is told to stop rather than being left running.
expect(calls[0].signal.aborted).toBe(true);
});
it("fails fast with no side effects when the graph is already aborted at entry", async () => {
const liveTask = mergeReadyTask();
const { store, executor } = executorFor(liveTask);
const { requester } = pendingMergeRequester();
executor.setMergeRequester(requester);
const controller = new AbortController();
controller.abort();
const result = await executor
.createAuthoritativeWorkflowPrimitives({ autoMerge: true })
.requestMerge(mergeCtx(controller.signal), liveTask);
expect(result).toMatchObject({ outcome: "failure", value: "merge-cancelled" });
// An abandoned walk must not enqueue a merge or mutate the boundary row.
expect(requester).not.toHaveBeenCalled();
expect(store.moveTask).not.toHaveBeenCalled();
});
it("still passes a live signal to the merger when the graph provides none", async () => {
const liveTask = mergeReadyTask();
const { executor } = executorFor(liveTask);
const inReview = { ...liveTask, column: "in-review" };
const requester = vi.fn(async () => ({ task: inReview, merged: true, noOp: true, mergeConfirmed: true }));
executor.setMergeRequester(requester);
const result = await executor
.createAuthoritativeWorkflowPrimitives({ autoMerge: true })
.requestMerge(mergeCtx(undefined), liveTask);
expect(result).toMatchObject({ outcome: "success" });
// The timeout controller's signal survives the AbortSignal.any linking.
expect(requester).toHaveBeenCalledWith("FN-CANCEL", expect.objectContaining({ signal: expect.any(AbortSignal) }));
});
});
describe("legacy merge seam", () => {
it("collapses immediately when the graph aborts mid-merge", async () => {
const liveTask = mergeReadyTask();
const { executor } = executorFor(liveTask);
const { requester, calls } = pendingMergeRequester();
executor.setMergeRequester(requester);
const controller = new AbortController();
const pending = executor
.createAuthoritativeWorkflowSeams({ autoMerge: true })
.merge(liveTask, {}, controller.signal);
await vi.waitFor(() => expect(requester).toHaveBeenCalled());
controller.abort();
await expect(pending).resolves.toMatchObject({ outcome: "failure", value: "merge-cancelled" });
expect(calls[0].signal.aborted).toBe(true);
});
it("fails fast with no side effects when already aborted at entry", async () => {
const liveTask = mergeReadyTask();
const { store, executor } = executorFor(liveTask);
const { requester } = pendingMergeRequester();
executor.setMergeRequester(requester);
const controller = new AbortController();
controller.abort();
const result = await executor
.createAuthoritativeWorkflowSeams({ autoMerge: true })
.merge(liveTask, {}, controller.signal);
expect(result).toMatchObject({ outcome: "failure", value: "merge-cancelled" });
expect(requester).not.toHaveBeenCalled();
expect(store.moveTask).not.toHaveBeenCalled();
});
});
describe("signal plumbing", () => {
it("carries the graph signal onto the primitive context", () => {
const controller = new AbortController();
const ctx = primitiveNodeContext(
{ runId: "run-1", taskId: "FN-1", workflowId: "coding" },
{ id: "merge", kind: "prompt" },
{},
controller.signal,
);
expect(ctx.signal).toBe(controller.signal);
});
it("forwards the node execution signal through the merge runner to the primitive", async () => {
const controller = new AbortController();
const requestMerge = vi.fn(async () => ({ outcome: "success", data: { status: "merged" } }));
const handler = createMergeAttemptHandler({
primitives: { requestMerge, audit: vi.fn() },
seams: { merge: vi.fn() },
buildPrimitiveContext: (node, ctx, attempt) =>
primitiveNodeContext({ runId: "r", taskId: "FN-1", workflowId: "w" }, node, { attempt }, ctx.signal),
});
await handler({ id: "merge", kind: "prompt" }, {
task: { id: "FN-1" },
settings: undefined,
context: {},
signal: controller.signal,
});
expect(requestMerge).toHaveBeenCalledWith(
expect.objectContaining({ signal: controller.signal }),
expect.anything(),
);
});
it("forwards the node execution signal to the legacy seam when no primitives are wired", async () => {
const controller = new AbortController();
const merge = vi.fn(async () => ({ outcome: "success" }));
const handler = createMergeAttemptHandler({
primitives: undefined,
seams: { merge },
buildPrimitiveContext: () => ({ run: {}, node: {} }),
});
await handler({ id: "merge", kind: "prompt" }, {
task: { id: "FN-1" },
settings: undefined,
context: {},
signal: controller.signal,
});
expect(merge).toHaveBeenCalledWith(expect.anything(), expect.anything(), controller.signal);
});
});
describe("classification", () => {
/* A cancellation routed into bounded auto-merge retry would re-request the merge the
operator just cancelled. `merge-cancelled` must stay a plain failure the graph's own
abort handling owns — never `transient-failure` (retry) or `merge-failed`. */
it("does not classify a cancellation as a retryable merge failure", () => {
expect(classifyMergePrimitiveResult(undefined, "merge-cancelled", "failure")).toEqual({
outcome: "failure",
value: "merge-cancelled",
});
});
});
});

46
packages/engine/src/executor.ts
View File

@@ -6476,6 +6476,13 @@ export class TaskExecutor {
if (!this.mergeRequester) {
return { outcome: "failure", value: "merge-unavailable", data: { status: "failed", reason: "merge-unavailable" } };
}
/*
FNXC:WorkflowCancellation 2026-07-15-10:42:
Fail fast on an already-cancelled walk BEFORE any side effect. `ensureWorkflowMergeBoundaryTask` mutates the task row and the requester enqueues a real merge; neither may run for a walk the engine has already abandoned. `merge-cancelled` is deliberately not `data.status: "failed"` — `classifyMergeFailure` would read an unknown reason as `merge-failed` and route a cancellation into bounded auto-merge retry.
*/
if (ctx.signal?.aborted) {
return { outcome: "failure", value: "merge-cancelled" };
}
const mergeTask = await this.ensureWorkflowMergeBoundaryTask(task, {
reason: "workflow-merge-boundary",
nodeId: ctx.node.node.id,
@@ -6516,8 +6523,13 @@ export class TaskExecutor {
data: { status: "failed", reason: "implementation-incomplete" },
};
}
/*
FNXC:WorkflowCancellation 2026-07-15-10:42:
The timeout bounds a wedged merge queue; it is NOT the cancellation path. `ctx.signal` (graph abort) is linked in via `AbortSignal.any` so a hard-cancel collapses the merge node immediately instead of after the full timeout, and is raced separately so the walk returns rather than waiting on a requester that may not settle on abort. Keep both signals live: dropping the timeout re-strands the walk behind a wedged queue, dropping the cancel link restores the 30-minute stall.
*/
const GRAPH_MERGE_TIMEOUT_MS = 30 * 60 * 1000;
const controller = new AbortController();
const mergeSignal = ctx.signal ? AbortSignal.any([ctx.signal, controller.signal]) : controller.signal;
let timeoutHandle: ReturnType<typeof setTimeout> | undefined;
const timeout = new Promise<"timeout">((resolve) => {
timeoutHandle = setTimeout(() => {
@@ -6526,8 +6538,18 @@ export class TaskExecutor {
}, GRAPH_MERGE_TIMEOUT_MS);
timeoutHandle.unref?.();
});
let onGraphAbort: (() => void) | undefined;
const cancelled = new Promise<"cancelled">((resolve) => {
if (!ctx.signal) return;
onGraphAbort = () => resolve("cancelled");
ctx.signal.addEventListener("abort", onGraphAbort, { once: true });
});
try {
const result = await Promise.race([this.mergeRequester(mergeTask.id, { signal: controller.signal }), timeout]);
const result = await Promise.race([this.mergeRequester(mergeTask.id, { signal: mergeSignal }), timeout, cancelled]);
if (result === "cancelled") {
executorLog.warn(`${mergeTask.id}: workflow merge primitive cancelled by graph abort`);
return { outcome: "failure", value: "merge-cancelled" };
}
if (result === "timeout") {
executorLog.warn(`${mergeTask.id}: workflow merge primitive timed out after ${GRAPH_MERGE_TIMEOUT_MS}ms`);
return { outcome: "failure", value: "merge-timeout", data: { status: "timeout" } };
@@ -6574,6 +6596,8 @@ export class TaskExecutor {
};
} finally {
if (timeoutHandle) clearTimeout(timeoutHandle);
// FNXC:WorkflowCancellation 2026-07-15-10:42: the graph signal outlives this node; leaving the listener attached leaks one per merge attempt across a retry loop.
if (onGraphAbort) ctx.signal?.removeEventListener("abort", onGraphAbort);
await logAudit(mergeTask.id, {
type: "merge-requested",
message: `Workflow node ${ctx.node.node.id} requested merge`,
@@ -6781,10 +6805,14 @@ export class TaskExecutor {
await this.handoffTaskToReview(live, "workflow-graph-review-handoff");
return { outcome: "success", value: "in-review" };
},
merge: async (seamTask) => {
merge: async (seamTask, _context, signal) => {
if (!this.mergeRequester) {
return { outcome: "failure", value: "merge-unavailable" };
}
// FNXC:WorkflowCancellation 2026-07-15-10:42: fail fast before the boundary-task mutation and the merge request — an abandoned walk must not enqueue a merge. Mirrors the `requestMerge` primitive.
if (signal?.aborted) {
return { outcome: "failure", value: "merge-cancelled" };
}
const mergeTask = await this.ensureWorkflowMergeBoundaryTask(seamTask, {
reason: "workflow-merge-boundary",
nodeId: "legacy-merge-seam",
@@ -6804,14 +6832,25 @@ export class TaskExecutor {
// Bound the wait: a wedged merge queue must not strand the graph walk
// holding the routing claim. On timeout the run fails cleanly and the
// task is parked for human review; the queue can still finish later.
// FNXC:WorkflowCancellation 2026-07-15-10:42: the timeout is the wedged-queue bound, `signal` is the cancellation path — both must stay live. See the `requestMerge` primitive for the stall this prevents.
const GRAPH_MERGE_TIMEOUT_MS = 30 * 60 * 1000;
let timeoutHandle: ReturnType<typeof setTimeout> | undefined;
const timeout = new Promise<"timeout">((resolve) => {
timeoutHandle = setTimeout(() => resolve("timeout"), GRAPH_MERGE_TIMEOUT_MS);
timeoutHandle.unref?.();
});
let onGraphAbort: (() => void) | undefined;
const cancelled = new Promise<"cancelled">((resolve) => {
if (!signal) return;
onGraphAbort = () => resolve("cancelled");
signal.addEventListener("abort", onGraphAbort, { once: true });
});
try {
const result = await Promise.race([this.mergeRequester(mergeTask.id), timeout]);
const result = await Promise.race([this.mergeRequester(mergeTask.id, signal ? { signal } : undefined), timeout, cancelled]);
if (result === "cancelled") {
executorLog.warn(`${mergeTask.id}: graph merge seam cancelled by graph abort`);
return { outcome: "failure", value: "merge-cancelled" };
}
if (result === "timeout") {
executorLog.warn(`${mergeTask.id}: graph merge seam timed out after ${GRAPH_MERGE_TIMEOUT_MS}ms`);
return { outcome: "failure", value: "merge-timeout" };
@@ -6822,6 +6861,7 @@ export class TaskExecutor {
return { outcome: "failure", value: result.reason ?? result.error ?? "merge-failed" };
} finally {
if (timeoutHandle) clearTimeout(timeoutHandle);
if (onGraphAbort) signal?.removeEventListener("abort", onGraphAbort);
}
},
schedule: async () => ({ outcome: "success" }),

10
packages/engine/src/runtime-primitives.ts
View File

@@ -38,6 +38,13 @@ export interface WorkflowRuntimeNodeContext {
export interface WorkflowPrimitiveContext {
run: WorkflowRuntimeRunContext;
node: WorkflowRuntimeNodeContext;
/*
FNXC:WorkflowCancellation 2026-07-15-10:42:
Graph cancellation must reach long-running primitives, not just node handlers. Before this existed, a hard-cancel (user cancel, engine restart, pause/resume) aborted the graph controller but the in-flight `merge` primitive never saw it: it raced the merge only against its own 30-minute timeout, so the walk sat inside the merge node for the full timeout before discovering it had been cancelled half an hour earlier. The timeout's abort then killed the still-running AI merge mid-flight ("Manual-merge failed: Request was aborted"), which could land between merger-ai's `worktree: null` write and `mergeConfirmed`, stranding the card as `no-worktree-no-merge-confirmed`.
Mirrors `WorkflowNodeExecutionContext.signal` (workflow-graph-executor.ts) and is threaded by `primitiveContextForNode`. Undefined on the sequential/uncancellable path. A primitive that can block on I/O for more than a few seconds MUST honor it — link it into any local timeout controller via `AbortSignal.any` rather than replacing it, so both cancellation and the timeout stay live.
*/
signal?: AbortSignal;
}
export interface RuntimePrimitiveResult<TValue = unknown> {
@@ -212,6 +219,8 @@ export function primitiveNodeContext(
run: WorkflowRuntimeRunContext,
node: WorkflowRuntimeNodeContext["node"],
extras: Omit<WorkflowRuntimeNodeContext, "node"> = {},
/** FNXC:WorkflowCancellation 2026-07-15-10:42: graph cancellation signal — see {@link WorkflowPrimitiveContext.signal}. */
signal?: AbortSignal,
): WorkflowPrimitiveContext {
return {
run,
@@ -219,6 +228,7 @@ export function primitiveNodeContext(
...extras,
node,
},
signal,
};
}

17
packages/engine/src/workflow-node-handlers.ts
View File

@@ -72,7 +72,8 @@ export interface WorkflowLegacySeams {
execute: (task: TaskDetail, context: Record<string, unknown>) => Promise<WorkflowNodeResult>;
review: (task: TaskDetail, context: Record<string, unknown>) => Promise<WorkflowNodeResult>;
"review-handoff"?: (task: TaskDetail, context: Record<string, unknown>) => Promise<WorkflowNodeResult>;
merge: (task: TaskDetail, context: Record<string, unknown>) => Promise<WorkflowNodeResult>;
/** FNXC:WorkflowCancellation 2026-07-15-10:42: `signal` carries graph cancellation into the legacy merge seam, which — like the merge primitive — otherwise blocks on its own 30-minute timeout and cannot observe a hard-cancel. Optional so seam fakes need not implement it. */
merge: (task: TaskDetail, context: Record<string, unknown>, signal?: AbortSignal) => Promise<WorkflowNodeResult>;
schedule: (task: TaskDetail, context: Record<string, unknown>) => Promise<WorkflowNodeResult>;
/**
* Step-inversion (KTD-2/KTD-4, U3): run exactly the foreach-active step inside
@@ -216,11 +217,16 @@ export type WorkflowCustomNodeRunner = (
context: Record<string, unknown>,
) => Promise<WorkflowNodeResult>;
/*
FNXC:WorkflowCancellation 2026-07-15-10:42:
`signal` is the graph walk's cancellation signal, taken from the node's own `WorkflowNodeExecutionContext.signal`. It was previously dropped here, leaving primitives unable to observe a graph abort — see {@link WorkflowPrimitiveContext.signal} for the 30-minute merge stall that caused. Every call site must forward its exec-context signal; a primitive that receives `undefined` cannot be cancelled.
*/
function primitiveContextForNode(
node: WorkflowIrNode,
task: TaskDetail,
context: Record<string, unknown>,
attempt?: number,
signal?: AbortSignal,
): WorkflowPrimitiveContext {
return primitiveNodeContext(
{
@@ -241,6 +247,7 @@ function primitiveContextForNode(
? context["workflow:effective-principal-id"]
: undefined,
},
signal,
);
}
@@ -351,7 +358,7 @@ export function createPrimitivePromptLikeHandler(
node.id,
);
const result = await primitives.runTaskStep(
primitiveContextForNode(node, context.task, context.context),
primitiveContextForNode(node, context.task, context.context, undefined, context.signal),
context.task,
active.stepIndex,
);
@@ -367,7 +374,7 @@ export function createPrimitivePromptLikeHandler(
}
if (seam) {
context.context[SEAM_GOVERNING_NODE_CONTEXT_KEY] = node.id;
const primitiveCtx = primitiveContextForNode(node, context.task, context.context);
const primitiveCtx = primitiveContextForNode(node, context.task, context.context, undefined, context.signal);
if (seam === "planning") {
const result = await primitives.runPlanningSession(primitiveCtx, context.task);
return { outcome: result.outcome, value: result.value, contextPatch: result.contextPatch };
@@ -546,7 +553,7 @@ export function createPrimitiveStepReviewHandler(primitives: WorkflowRuntimePrim
let primitivePatch: Record<string, unknown> | undefined;
for (let attempt = 0; attempt <= STEP_REVIEW_UNAVAILABLE_RETRY_CAP; attempt++) {
const primitiveResult = await primitives.runReview(
primitiveContextForNode(node, ctx.task, ctx.context, attempt + 1),
primitiveContextForNode(node, ctx.task, ctx.context, attempt + 1, ctx.signal),
ctx.task,
{
type: config.type,
@@ -679,7 +686,7 @@ export function createDefaultNodeHandlers(
primitives: deps?.primitives,
seams,
buildPrimitiveContext: (node, ctx, attempt) =>
primitiveContextForNode(node, ctx.task, ctx.context, attempt),
primitiveContextForNode(node, ctx.task, ctx.context, attempt, ctx.signal),
}),
"manual-merge-hold": async () => ({ outcome: "failure", value: "manual-required" }),
"retry-backoff": async () => ({ outcome: "success" }),

2
packages/engine/src/workflow-node-runners/merge-runner.ts
View File

@@ -25,7 +25,7 @@ Merge-attempt behavior is isolated behind a runner factory so the graph handler
export function createMergeAttemptHandler(deps: MergeAttemptRunnerDeps): WorkflowNodeHandler {
return async (node, ctx) => {
if (!deps.primitives) {
return deps.seams.merge(ctx.task, ctx.context);
return deps.seams.merge(ctx.task, ctx.context, ctx.signal);
}
const attempt = typeof ctx.context["workflow:work-item-attempt"] === "number"
? ctx.context["workflow:work-item-attempt"]