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>
This commit is contained in:
gsxdsm
@@ -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();
|
||||
|
||||
@@ -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)}`,
|
||||
|
||||
Reference in New Issue
Block a user