feat(fusion): make auto-merge queue priority-aware

Triage and the todo→in-progress scheduler already sorted by priority
(urgent→low, then createdAt ASC, then id ASC); the auto-merge queue
was strictly FIFO, so a backlogged low-priority task could merge
ahead of an urgent one. drainMergeQueue now picks the highest-
priority eligible task each iteration, and the four in-review sweeps
(startup, periodic, global unpause, engine unpause) sort by priority
before enqueueing so the single-item fast path also picks priority-
first. Picker is hardened against concurrent queue mutation by stop()
and pause-handler removal: it re-locates the chosen entry by id and
re-checks shuttingDown after awaiting getTask.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

.changeset/merge-queue-priority.md

@@ -0,0 +1,10 @@
+---
+"@runfusion/fusion": patch
+---
+
+Apply task priority across all Fusion scheduling paths so urgent work overtakes older low-priority work — including the merge queue, which previously merged tasks strictly FIFO.
+
+- The auto-merge queue now picks the highest-priority eligible task each iteration (`urgent → high → normal → low`, then `createdAt` ASC, then id ASC). Manual `onMerge` resolvers still run before auto-merges so awaited callers aren't starved.
+- Startup, periodic, global-unpause, and engine-unpause sweeps now sort their `listTasks` result by priority before enqueueing, so the first task picked up by `drainMergeQueue`'s single-item fast path is the highest-priority eligible one rather than the oldest. All four sweeps share a new `enqueueEligibleInReviewTasks` helper.
+- Hardened the picker against concurrent queue mutation: it now re-locates the chosen task via `indexOf` after awaiting `getTask`, so a `stop()` clear or pause-handler removal that lands during the await can't splice out the wrong sibling. Drain and picker both re-check `shuttingDown` after the awaits to avoid starting a merge whose queue entry was already cleared.
+- Triage and todo→in-progress scheduling already used the shared `sortTasksByPriorityThenAgeAndId` comparator and continue to apply dependency, overlap, and worktree constraints after the priority sort.

packages/engine/src/__tests__/project-engine.test.ts

@@ -1014,6 +1014,248 @@ describe("ProjectEngine shutdown merge handling", () => {
   });
 });
 
+describe("ProjectEngine merge queue priority ordering", () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+  });
+
+  it("merges higher-priority tasks before lower-priority ones regardless of enqueue order", async () => {
+    const mockStore = createMockStore({ ...baseSettings, autoMerge: true });
+    const tasksById: Record<string, Record<string, unknown>> = {
+      "FN-low": {
+        id: "FN-low",
+        column: "in-review",
+        paused: false,
+        mergeRetries: 0,
+        status: null,
+        priority: "low",
+        createdAt: "2026-04-01T00:00:00.000Z",
+      },
+      "FN-urgent": {
+        id: "FN-urgent",
+        column: "in-review",
+        paused: false,
+        mergeRetries: 0,
+        status: null,
+        priority: "urgent",
+        createdAt: "2026-04-02T00:00:00.000Z",
+      },
+      "FN-normal": {
+        id: "FN-normal",
+        column: "in-review",
+        paused: false,
+        mergeRetries: 0,
+        status: null,
+        priority: "normal",
+        createdAt: "2026-04-03T00:00:00.000Z",
+      },
+    };
+    mockStore.store.getTask.mockImplementation(async (id: string) => tasksById[id] ?? null);
+    mocks.currentStore = mockStore.store;
+
+    const mergeOrder: string[] = [];
+    mocks.aiMergeTask.mockImplementation(async (...args: unknown[]) => {
+      mergeOrder.push(args[2] as string);
+      return { merged: true } as never;
+    });
+
+    const engine = createEngine();
+    await engine.start();
+
+    // Enqueue lowest priority first, urgent last. Priority-aware dequeue must
+    // still surface FN-urgent before FN-normal regardless of enqueue order.
+    engine.enqueueMerge("FN-low");
+    engine.enqueueMerge("FN-normal");
+    engine.enqueueMerge("FN-urgent");
+
+    await vi.waitFor(() => {
+      expect(mergeOrder).toHaveLength(3);
+    });
+
+    // FN-low may merge first if drainMergeQueue picked it up before the other
+    // enqueues landed (single-item fast path). The contract is that once 2+
+    // tasks are queued together, the higher-priority one wins — so FN-urgent
+    // (enqueued last) must merge before FN-normal (enqueued before it).
+    const urgentIdx = mergeOrder.indexOf("FN-urgent");
+    const normalIdx = mergeOrder.indexOf("FN-normal");
+    expect(urgentIdx).toBeGreaterThanOrEqual(0);
+    expect(normalIdx).toBeGreaterThanOrEqual(0);
+    expect(urgentIdx).toBeLessThan(normalIdx);
+
+    await engine.stop();
+  });
+
+  it("startup sweep merges higher-priority tasks first even though listTasks returns oldest-first", async () => {
+    const mockStore = createMockStore({ ...baseSettings, autoMerge: true });
+    // Tasks returned in createdAt ASC order (matches store.listTasks contract).
+    // Priority order is interleaved so a naive iteration would merge FN-low
+    // first; priority-aware sorting must reorder to urgent → normal → low.
+    const sweptTasks = [
+      {
+        id: "FN-low",
+        column: "in-review",
+        paused: false,
+        mergeRetries: 0,
+        status: null,
+        priority: "low",
+        createdAt: "2026-04-01T00:00:00.000Z",
+      },
+      {
+        id: "FN-urgent",
+        column: "in-review",
+        paused: false,
+        mergeRetries: 0,
+        status: null,
+        priority: "urgent",
+        createdAt: "2026-04-02T00:00:00.000Z",
+      },
+      {
+        id: "FN-normal",
+        column: "in-review",
+        paused: false,
+        mergeRetries: 0,
+        status: null,
+        priority: "normal",
+        createdAt: "2026-04-03T00:00:00.000Z",
+      },
+    ];
+    const tasksById: Record<string, Record<string, unknown>> = Object.fromEntries(
+      sweptTasks.map((t) => [t.id, t]),
+    );
+    mockStore.store.listTasks.mockResolvedValue(sweptTasks);
+    mockStore.store.getTask.mockImplementation(async (id: string) => tasksById[id] ?? null);
+    mocks.currentStore = mockStore.store;
+
+    const mergeOrder: string[] = [];
+    mocks.aiMergeTask.mockImplementation(async (...args: unknown[]) => {
+      mergeOrder.push(args[2] as string);
+      return { merged: true } as never;
+    });
+
+    const engine = createEngine();
+    await engine.start();
+
+    await vi.waitFor(() => {
+      expect(mergeOrder).toHaveLength(3);
+    });
+
+    expect(mergeOrder).toEqual(["FN-urgent", "FN-normal", "FN-low"]);
+
+    await engine.stop();
+  });
+
+  // Direct unit-tests of pickNextMergeTaskId to exercise the multi-item
+  // priority path with concurrent queue mutations during getTask awaits.
+  // These are unreachable through enqueueMerge alone because the first
+  // enqueue always takes the single-item fast path.
+  it("picker falls back to next-priority task when the chosen one is removed from the queue during getTask awaits", async () => {
+    const mockStore = createMockStore({ ...baseSettings, autoMerge: true });
+    const tasksById: Record<string, Record<string, unknown>> = {
+      "FN-urgent": { id: "FN-urgent", column: "in-review", paused: false, mergeRetries: 0, status: null, priority: "urgent", createdAt: "2026-04-01T00:00:00.000Z" },
+      "FN-normal-a": { id: "FN-normal-a", column: "in-review", paused: false, mergeRetries: 0, status: null, priority: "normal", createdAt: "2026-04-02T00:00:00.000Z" },
+      "FN-normal-b": { id: "FN-normal-b", column: "in-review", paused: false, mergeRetries: 0, status: null, priority: "normal", createdAt: "2026-04-03T00:00:00.000Z" },
+    };
+
+    let releaseUrgent: (() => void) = () => {};
+    const urgentHeld = new Promise<void>((resolve) => {
+      releaseUrgent = resolve;
+    });
+    let urgentRequested = false;
+    mockStore.store.getTask.mockImplementation(async (id: string) => {
+      if (id === "FN-urgent") {
+        urgentRequested = true;
+        await urgentHeld;
+      }
+      return tasksById[id] ?? null;
+    });
+    mocks.currentStore = mockStore.store;
+
+    const engine = createEngine();
+    await engine.start();
+    const privateEngine = engine as unknown as {
+      mergeQueue: string[];
+      mergeActive: Set<string>;
+      pickNextMergeTaskId: (store: unknown) => Promise<string | undefined>;
+    };
+
+    privateEngine.mergeQueue = ["FN-urgent", "FN-normal-a", "FN-normal-b"];
+    privateEngine.mergeActive = new Set(["FN-urgent", "FN-normal-a", "FN-normal-b"]);
+
+    const pickPromise = privateEngine.pickNextMergeTaskId(mockStore.store);
+
+    await vi.waitFor(() => {
+      expect(urgentRequested).toBe(true);
+    });
+
+    // Simulate pause-handler removing FN-urgent mid-pick.
+    privateEngine.mergeQueue = privateEngine.mergeQueue.filter((id) => id !== "FN-urgent");
+    privateEngine.mergeActive.delete("FN-urgent");
+
+    releaseUrgent();
+    const chosen = await pickPromise;
+
+    // FN-urgent was yanked; picker must fall back to next-priority survivor.
+    // Both surviving tasks are "normal"; FN-normal-a wins by older createdAt.
+    expect(chosen).toBe("FN-normal-a");
+    expect(privateEngine.mergeQueue).toEqual(["FN-normal-b"]);
+
+    await engine.stop();
+  });
+
+  it("picker returns undefined when shutdown lands during getTask awaits", async () => {
+    const mockStore = createMockStore({ ...baseSettings, autoMerge: true });
+    const tasksById: Record<string, Record<string, unknown>> = {
+      "FN-a": { id: "FN-a", column: "in-review", paused: false, mergeRetries: 0, status: null, priority: "high", createdAt: "2026-04-01T00:00:00.000Z" },
+      "FN-b": { id: "FN-b", column: "in-review", paused: false, mergeRetries: 0, status: null, priority: "normal", createdAt: "2026-04-02T00:00:00.000Z" },
+    };
+
+    let release: (() => void) = () => {};
+    const held = new Promise<void>((resolve) => {
+      release = resolve;
+    });
+    let firstCallSeen = false;
+    mockStore.store.getTask.mockImplementation(async (id: string) => {
+      if (!firstCallSeen) {
+        firstCallSeen = true;
+        await held;
+      }
+      return tasksById[id] ?? null;
+    });
+    mocks.currentStore = mockStore.store;
+
+    const engine = createEngine();
+    await engine.start();
+    const privateEngine = engine as unknown as {
+      mergeQueue: string[];
+      mergeActive: Set<string>;
+      shuttingDown: boolean;
+      pickNextMergeTaskId: (store: unknown) => Promise<string | undefined>;
+    };
+
+    privateEngine.mergeQueue = ["FN-a", "FN-b"];
+    privateEngine.mergeActive = new Set(["FN-a", "FN-b"]);
+
+    const pickPromise = privateEngine.pickNextMergeTaskId(mockStore.store);
+
+    await vi.waitFor(() => {
+      expect(firstCallSeen).toBe(true);
+    });
+
+    // Simulate shutdown while picker is awaiting getTask.
+    privateEngine.shuttingDown = true;
+    privateEngine.mergeQueue = [];
+
+    release();
+    const chosen = await pickPromise;
+
+    expect(chosen).toBeUndefined();
+
+    // Reset so engine.stop() teardown runs cleanly.
+    privateEngine.shuttingDown = false;
+    await engine.stop();
+  });
+});
+
 describe("ProjectEngine paused in-review auto-merge behavior", () => {
   beforeEach(() => {
     vi.clearAllMocks();

packages/engine/src/project-engine.ts

@@ -9,6 +9,7 @@ import type {
   ScheduledTask,
   AutomationRunResult,
 } from "@fusion/core";
+import { compareTasksByPriorityThenAgeAndId, sortTasksByPriorityThenAgeAndId } from "@fusion/core";
 import { execFile } from "node:child_process";
 import { promisify } from "node:util";
 import { InProcessRuntime } from "./runtimes/in-process-runtime.js";
@@ -1017,6 +1018,60 @@ export class ProjectEngine {
     );
   }
 
+  /**
+   * Remove and return the highest-priority taskId from the merge queue.
+   * Ordering: priority (urgent→low), then createdAt ASC, then id ASC — matching
+   * the triage and scheduler comparators. Manual merges (onMerge resolvers) are
+   * preferred over auto-merges so awaited callers aren't starved by a flood of
+   * higher-priority auto-enqueues. IDs whose tasks can't be loaded fall back to
+   * FIFO order so they still drain.
+   */
+  private async pickNextMergeTaskId(store: TaskStore): Promise<string | undefined> {
+    if (this.mergeQueue.length === 0) return undefined;
+    // Fast path: with a single queued task there's nothing to reorder. Avoid an
+    // extra getTask round-trip (and keep callers that mock getTask once happy).
+    if (this.mergeQueue.length === 1) {
+      return this.mergeQueue.shift();
+    }
+
+    // Snapshot the queue before awaiting. While we await store.getTask for
+    // each id, stop() may clear mergeQueue and pause-handling may filter
+    // entries out — so we never trust positional indices afterwards.
+    const queueSnapshot = [...this.mergeQueue];
+    const entries: Array<{ taskId: string; task: Task | undefined; manual: boolean; order: number }> = [];
+    for (let i = 0; i < queueSnapshot.length; i++) {
+      const taskId = queueSnapshot[i]!;
+      const task = (await store.getTask(taskId).catch(() => undefined)) as Task | undefined;
+      entries.push({
+        taskId,
+        task,
+        manual: this.manualMergeResolvers.has(taskId),
+        order: i,
+      });
+    }
+
+    if (this.shuttingDown) return undefined;
+
+    entries.sort((a, b) => {
+      if (a.manual !== b.manual) return a.manual ? -1 : 1;
+      if (a.task && b.task) return compareTasksByPriorityThenAgeAndId(a.task, b.task);
+      if (a.task) return -1;
+      if (b.task) return 1;
+      return a.order - b.order;
+    });
+
+    // Find the highest-priority entry that is still in the live queue.
+    // Concurrent mutations (pause filter, stop) may have removed entries.
+    for (const entry of entries) {
+      const liveIndex = this.mergeQueue.indexOf(entry.taskId);
+      if (liveIndex !== -1) {
+        this.mergeQueue.splice(liveIndex, 1);
+        return entry.taskId;
+      }
+    }
+    return undefined;
+  }
+
   private internalEnqueueMerge(taskId: string): void {
     if (this.shuttingDown) return;
     if (this.mergeActive.has(taskId)) return;
@@ -1025,6 +1080,24 @@ export class ProjectEngine {
     void this.drainMergeQueue();
   }
 
+  /**
+   * Filter a sweep's listTasks() result to merge-eligible tasks, sort by
+   * priority (urgent → low, then createdAt ASC, then id ASC), and enqueue.
+   * Sorting before enqueue matters because each enqueue may immediately
+   * trigger drainMergeQueue's single-item fast path, so the first task
+   * pushed wins. listTasks returns createdAt ASC — without this sort an
+   * older low-priority task would start before a later urgent one.
+   */
+  private enqueueEligibleInReviewTasks(tasks: readonly Task[]): number {
+    const eligible = sortTasksByPriorityThenAgeAndId(
+      tasks.filter((t) => !t.paused && this.canMergeTask(t as any)) as Task[],
+    );
+    for (const t of eligible) {
+      this.internalEnqueueMerge(t.id);
+    }
+    return eligible.length;
+  }
+
   private async drainMergeQueue(): Promise<void> {
     if (this.mergeRunning) return;
     this.mergeRunning = true;
@@ -1034,7 +1107,11 @@ export class ProjectEngine {
       const cwd = this.config.workingDirectory;
 
       while (this.mergeQueue.length > 0 && !this.shuttingDown) {
-        const taskId = this.mergeQueue.shift()!;
+        const taskId = await this.pickNextMergeTaskId(store);
+        if (!taskId) break;
+        // pickNextMergeTaskId awaits store.getTask; re-check shutdown so we
+        // don't start a merge whose queue entry was cleared by stop().
+        if (this.shuttingDown) break;
         const manualResolver = this.manualMergeResolvers.get(taskId);
         try {
           // Manual merges (onMerge) skip auto-merge eligibility checks
@@ -1664,12 +1741,9 @@ export class ProjectEngine {
       if (!settings.autoMerge) return;
 
 
-      const eligible = tasks.filter((t) => !t.paused && this.canMergeTask(t as any));
-      if (eligible.length > 0) {
-        runtimeLog.log(`Auto-merge startup sweep: enqueueing ${eligible.length} task(s)`);
-        for (const t of eligible) {
-          this.internalEnqueueMerge(t.id);
-        }
+      const enqueued = this.enqueueEligibleInReviewTasks(tasks as Task[]);
+      if (enqueued > 0) {
+        runtimeLog.log(`Auto-merge startup sweep: enqueueing ${enqueued} task(s)`);
       }
     } catch (err: unknown) {
       runtimeLog.warn(
@@ -1688,14 +1762,7 @@ export class ProjectEngine {
         const settings = await store.getSettings();
         if (!settings.globalPause && !settings.enginePaused && settings.autoMerge) {
           const tasks = await store.listTasks({ column: "in-review" });
-          for (const t of tasks) {
-            if (t.paused) {
-              continue;
-            }
-            if (this.canMergeTask(t as any)) {
-              this.internalEnqueueMerge(t.id);
-            }
-          }
+          this.enqueueEligibleInReviewTasks(tasks as Task[]);
         }
       } catch (err: unknown) {
         runtimeLog.warn(
@@ -1771,14 +1838,7 @@ export class ProjectEngine {
         if (s.autoMerge) {
           try {
             const tasks = await store.listTasks({ column: "in-review" });
-            for (const t of tasks) {
-              if (t.paused) {
-                continue;
-              }
-              if (this.canMergeTask(t as any)) {
-                this.internalEnqueueMerge(t.id);
-              }
-            }
+            this.enqueueEligibleInReviewTasks(tasks as Task[]);
           } catch (err: unknown) {
             runtimeLog.warn(
               `Global unpause: failed to scan in-review tasks for auto-merge: ${err instanceof Error ? err.message : String(err)}`,
@@ -1816,14 +1876,7 @@ export class ProjectEngine {
         if (s.autoMerge) {
           try {
             const tasks = await store.listTasks({ column: "in-review" });
-            for (const t of tasks) {
-              if (t.paused) {
-                continue;
-              }
-              if (this.canMergeTask(t as any)) {
-                this.internalEnqueueMerge(t.id);
-              }
-            }
+            this.enqueueEligibleInReviewTasks(tasks as Task[]);
           } catch (err: unknown) {
             runtimeLog.warn(
               `Engine unpause: failed to scan in-review tasks for auto-merge: ${err instanceof Error ? err.message : String(err)}`,