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

feat(FN-4053): unify task ID allocation with store-owned distributed alloca

Browse Source 
Unifies task ID allocation under a single store-owned distributed allocator, removing redundant ID-generation logic from dashboard route handlers and simplifying the overall flow; Step 1 merges the allocation authority into the store, Step 3 removes the now-unnecessary mixed-path routing, and tests

Fusion-Task-Id: FN-4053
This commit is contained in:
Fusion
2026-05-11 21:38:25 -07:00
committed by gsxdsm
parent ce0e5f4ce6
commit 4404c6103e
12 changed files with 244 additions and 359 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
.changeset/unify-task-id-allocator.md Normal file
View File

@@ -0,0 +1,5 @@
---
"@runfusion/fusion": patch
---
Unify local task creation on the distributed task-ID allocator lifecycle and remove runtime reliance on `config.nextId` as an allocation counter. Local allocator state now self-heals on startup by reconciling to existing task IDs for each prefix.

2
docs/architecture.md
View File

@@ -671,6 +671,8 @@ A lease is recoverable only when there is **no active local executor session for
- `committedClusterTaskCount` from allocator state is the only authoritative cluster-wide committed-task count. Local task-row counts and ID suffix math are not authoritative.
- Mesh allocator write routes (`/api/mesh/task-ids/reserve|commit|abort`) return `503` when the coordinator node is unreachable; they never fall back to local-only cluster ID issuance.
- Cluster task creation now uses a strong-write reserve → create → replicate → commit/abort sequence.
- Ordinary local task creation (`TaskStore.createTask()`, duplicate, and refine flows) now allocates IDs through the same distributed reserve/commit/abort lifecycle owned by `TaskStore`.
- `POST /api/tasks` uses the store-owned allocator path for local creates rather than maintaining a separate route-local allocator implementation.
- `POST /api/tasks` reserves a distributed ID, creates the authoritative local task with that reserved ID, then POSTs authenticated replication payloads to peer nodes.
- Creation self-heals stale ID overlap state: if a reserved `FN-*` collides with an existing task (`Task ID already exists...` or replicated-create collision), the route aborts that reservation, cleans up partial local state, reserves the next ID, and retries up to a bounded limit.
- Replica apply uses `TaskStore.applyReplicatedTaskCreate(...)`, which is idempotent by task ID: replaying the same payload returns the existing task without creating duplicates.

7
docs/storage.md
View File

@@ -1,5 +1,12 @@
# Fusion Dashboard Storage Audit (FN-1202)
## Task-ID allocator authority and compatibility
- `distributed_task_id_state` is the authoritative local task-ID allocator state. `nextSequence` is the active high-water mark used for local ID reservations.
- `distributed_task_id_reservations` tracks reserve/commit/abort lifecycle entries. Aborted/expired reservations are burned and never reissued.
- `config.nextId` is retained only as a legacy compatibility field and optional seed source; runtime task creation no longer mutates it as allocator truth.
- Startup allocator reconciliation bumps each active prefix sequence to `max(current nextSequence, max(existing task suffix)+1)` across live + archived tasks to self-heal stale allocator drift.
## 1) Summary
- **localStorage keys in runtime dashboard code:** **20**

16
packages/core/src/__tests__/store-activity.test.ts
View File

@@ -562,28 +562,20 @@ describe("TaskStore", () => {
}
});
it("logs allocateId disk sync failures while preserving task creation", async () => {
it("creates tasks through distributed allocation without config.json sync dependency", async () => {
const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {});
const storeAny = store as any;
const originalConfigPath = storeAny.configPath;
storeAny.configPath = join(rootDir, ".fusion", "missing-sync", "config.json");
try {
const task = await store.createTask({ description: "allocate despite sync failure" });
const task = await store.createTask({ description: "allocate without config sync" });
expect(task.id).toBe("FN-001");
const warningCall = warnSpy.mock.calls.find(
(call) => typeof call[0] === "string" && call[0].includes("[task-store] Backward-compat config.json sync failed after ID allocation"),
(call) => typeof call[0] === "string" && call[0].includes("after ID allocation"),
);
expect(warningCall).toBeDefined();
const [, context] = warningCall as [string, Record<string, unknown>];
expect(context).toMatchObject({
phase: "allocateId:disk-sync",
configPath: join(rootDir, ".fusion", "missing-sync", "config.json"),
taskId: task.id,
});
expect(typeof context.error).toBe("string");
expect(warningCall).toBeUndefined();
} finally {
storeAny.configPath = originalConfigPath;
warnSpy.mockRestore();

47
packages/core/src/__tests__/store-create.test.ts
View File

@@ -707,6 +707,53 @@ describe("TaskStore", () => {
describe("distributed task-id allocator seam", () => {
it("commits allocator reservations for createTask, duplicateTask, and refineTask", async () => {
const created = await store.createTask({ description: "created with allocator" });
const duplicate = await store.duplicateTask(created.id);
await store.moveTask(created.id, "todo");
await store.moveTask(created.id, "in-progress");
await store.moveTask(created.id, "in-review");
await store.moveTask(created.id, "done");
const refined = await store.refineTask(created.id, "refine this");
const reservationRows = store
.getDatabase()
.prepare("SELECT taskId, status FROM distributed_task_id_reservations WHERE taskId IN (?, ?, ?) ORDER BY taskId")
.all(created.id, duplicate.id, refined.id) as Array<{ taskId: string; status: string }>;
expect(reservationRows).toEqual([
{ taskId: created.id, status: "committed" },
{ taskId: duplicate.id, status: "committed" },
{ taskId: refined.id, status: "committed" },
]);
});
it("keeps IDs collision-free across mixed store and direct reservation creates", async () => {
// Regression for FN-4053: before unifying local allocation, store.createTask()
// used config.nextId while direct distributed reservations advanced
// distributed_task_id_state. Interleaving both paths could reuse IDs.
const first = await store.createTask({ description: "first via store" });
const allocator = store.getDistributedTaskIdAllocator();
const reservation = await allocator.reserveDistributedTaskId({ prefix: "FN", nodeId: "node-a" });
const second = await store.createTaskWithReservedId(
{ description: "second via direct reservation" },
{ taskId: reservation.taskId },
);
await allocator.commitDistributedTaskIdReservation({
reservationId: reservation.reservationId,
nodeId: "node-a",
});
const third = await store.createTask({ description: "third via store" });
expect(first.id).toBe("FN-001");
expect(second.id).toBe("FN-002");
expect(third.id).toBe("FN-003");
expect(third.id).not.toBe(second.id);
});
it("returns a stable allocator instance", () => {
const first = store.getDistributedTaskIdAllocator();
const second = store.getDistributedTaskIdAllocator();

5
packages/core/src/__tests__/store-resilience.test.ts
View File

@@ -134,11 +134,10 @@ describe("TaskStore", () => {
const sortedIds = [...ids].sort();
expect(sortedIds).toEqual(["FN-001", "FN-002", "FN-003", "FN-004", "FN-005"]);
// config.json should be valid JSON with nextId = 6
// config.json should still be valid JSON after concurrent task creation
const configPath = join(rootDir, ".fusion", "config.json");
const raw = await readFile(configPath, "utf-8");
const config = JSON.parse(raw);
expect(config.nextId).toBe(6);
expect(() => JSON.parse(raw)).not.toThrow();
// No .tmp files left behind
const haiDir = join(rootDir, ".fusion");

6
packages/core/src/__tests__/store-settings.test.ts
View File

@@ -2054,7 +2054,7 @@ describe("TaskStore", () => {
expect(task.id).toBe("PROJ-001");
});
it("prefix change mid-stream continues sequence", async () => {
it("prefix change mid-stream starts a fresh per-prefix sequence", async () => {
const t1 = await harness.store().createTask({ description: "First" });
const t2 = await harness.store().createTask({ description: "Second" });
expect(t1.id).toBe("FN-001");
@@ -2062,7 +2062,7 @@ describe("TaskStore", () => {
await harness.store().updateSettings({ taskPrefix: "PROJ" });
const t3 = await harness.store().createTask({ description: "Third" });
expect(t3.id).toBe("PROJ-003");
expect(t3.id).toBe("PROJ-001");
});
it("listTasks returns tasks regardless of prefix", async () => {
@@ -2072,7 +2072,7 @@ describe("TaskStore", () => {
const tasks = await harness.store().listTasks();
expect(tasks).toHaveLength(2);
expect(tasks.map((t) => t.id).sort()).toEqual(["FN-001", "PROJ-002"]);
expect(tasks.map((t) => t.id).sort()).toEqual(["FN-001", "PROJ-001"]);
});
it("supports pagination with limit and offset", async () => {

17
packages/core/src/distributed-task-id.ts
View File

@@ -21,6 +21,14 @@ export interface DistributedTaskIdAllocator {
getDistributedTaskIdState(input: DistributedTaskIdStateInput): Promise<DistributedTaskIdStateResult>;
}
export function resolveLocalNodeId(
nodes: Array<{ id: string; type: string }> | undefined,
fallback = "local",
): string {
const localNode = nodes?.find((node) => node.type === "local");
return localNode?.id ?? fallback;
}
export class DistributedTaskIdError extends Error {
constructor(
message: string,
@@ -147,11 +155,18 @@ export function createDistributedTaskIdAllocator(db: Database): DistributedTaskI
};
probeTable("tasks");
probeTable("archivedTasks");
const nowIso = new Date().toISOString();
db.prepare(
`INSERT OR IGNORE INTO distributed_task_id_state (
prefix, nextSequence, committedClusterTaskCount, lastCommittedTaskId, updatedAt
) VALUES (?, ?, 0, NULL, ?)`
).run(prefix, seedSequence, new Date().toISOString());
).run(prefix, seedSequence, nowIso);
db.prepare(
`UPDATE distributed_task_id_state
SET nextSequence = MAX(nextSequence, ?),
updatedAt = ?
WHERE prefix = ?`
).run(seedSequence, nowIso, prefix);
};
return {

1
packages/core/src/index.ts
View File

@@ -98,6 +98,7 @@ export {
export {
createDistributedTaskIdAllocator,
formatDistributedTaskId,
resolveLocalNodeId,
DistributedTaskIdError,
} from "./distributed-task-id.js";
export type { DistributedTaskIdAllocator } from "./distributed-task-id.js";

287
packages/core/src/store.ts
View File

@@ -28,7 +28,7 @@ import { validateNodeOverrideChange } from "./node-override-guard.js";
import { sanitizeTitle } from "./ai-summarize.js";
import { assertProjectRootDir } from "./project-root-guard.js";
import { generateTaskLineageId, normalizeTaskCommitAssociation } from "./task-lineage.js";
import { createDistributedTaskIdAllocator, type DistributedTaskIdAllocator } from "./distributed-task-id.js";
import { createDistributedTaskIdAllocator, resolveLocalNodeId, type DistributedTaskIdAllocator } from "./distributed-task-id.js";
import {
buildBootstrapPrompt,
replicationCollisionError,
@@ -2146,39 +2146,57 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
}
}
private async allocateId(): Promise<string> {
// Use withConfigLock to ensure the entire ID allocation + config sync is serialized
return this.withConfigLock(async () => {
const id = this.db.transaction(() => {
const row = this.db.prepare("SELECT nextId, settings FROM config WHERE id = 1").get() as unknown as { nextId: number; settings: string | null } | undefined;
const settings = fromJson<Settings>(row?.settings ?? null);
const prefix = settings?.taskPrefix || "KB";
const nextId = row?.nextId || 1;
const taskId = `${prefix}-${String(nextId).padStart(3, "0")}`;
this.db.prepare("UPDATE config SET nextId = ? WHERE id = 1").run(nextId + 1);
this.db.bumpLastModified();
return taskId;
}); // Database.transaction() directly executes and returns the result
async resolveLocalNodeIdForTaskAllocation(): Promise<string> {
if (process.env.VITEST === "true") {
return "local";
}
const central = new CentralCore();
await central.init();
try {
const nodes = await central.listNodes();
return resolveLocalNodeId(nodes.map((node) => ({ id: node.id, type: node.type })));
} catch {
return "local";
} finally {
await central.close();
}
}
// Sync config.json to disk for backward compatibility.
// Use readConfigFast() to avoid the expensive listWorkflowSteps() query.
try {
const config = this.readConfigFast();
const tmpPath = this.configPath + ".tmp";
await writeFile(tmpPath, JSON.stringify(config, null, 2));
await rename(tmpPath, this.configPath);
} catch (err) {
// Non-fatal: SQLite is the primary store
storeLog.warn("Backward-compat config.json sync failed after ID allocation", {
phase: "allocateId:disk-sync",
configPath: this.configPath,
taskId: id,
error: err instanceof Error ? err.message : String(err),
});
}
return id;
private async createTaskWithDistributedReservation(
input: TaskCreateInput,
options?: {
onSummarize?: (description: string) => Promise<string | null>;
settings?: { autoSummarizeTitles?: boolean };
createTaskWithId?: (taskId: string) => Promise<Task>;
},
): Promise<Task> {
const settings = await this.getSettingsFast();
const prefix = (settings.taskPrefix || "KB").trim().toUpperCase();
const allocator = this.getDistributedTaskIdAllocator();
const nodeId = await this.resolveLocalNodeIdForTaskAllocation();
const reservation = await allocator.reserveDistributedTaskId({
prefix,
nodeId,
});
let createdTask: Task | null = null;
try {
createdTask = options?.createTaskWithId
? await options.createTaskWithId(reservation.taskId)
: await this.createTaskWithReservedId(input, { taskId: reservation.taskId });
await allocator.commitDistributedTaskIdReservation({
reservationId: reservation.reservationId,
nodeId,
});
return createdTask;
} catch (error) {
await allocator.abortDistributedTaskIdReservation({
reservationId: reservation.reservationId,
nodeId,
reason: "failed-create",
}).catch(() => undefined);
throw error;
}
}
private taskDir(id: string): string {
@@ -2347,26 +2365,18 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
throw new Error("Description is required and cannot be empty");
}
const id = await this.allocateId();
// Validate that task doesn't depend on itself
if (input.dependencies?.includes(id)) {
throw new Error(`Task ${id} cannot depend on itself`);
}
// Determine if we should try to summarize the title
const title = input.title?.trim() || undefined;
const shouldSummarize =
!title && // Only if no title provided
input.description.length > 200 && // Only if description is long enough
(input.summarize === true || // Explicit request
options?.settings?.autoSummarizeTitles === true); // Auto-enabled
!title &&
input.description.length > 200 &&
(input.summarize === true || options?.settings?.autoSummarizeTitles === true);
// Determine enabledWorkflowSteps: explicit input takes precedence, otherwise auto-apply default-on steps
let resolvedWorkflowSteps: string[] | undefined = input.enabledWorkflowSteps?.length
? await this.resolveEnabledWorkflowSteps(input.enabledWorkflowSteps)
: undefined;
// When enabledWorkflowSteps is not provided at all (undefined), auto-apply default-on workflow steps
if (input.enabledWorkflowSteps === undefined) {
try {
const allSteps = await this.listWorkflowSteps();
@@ -2377,7 +2387,6 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
resolvedWorkflowSteps = defaultOnSteps;
}
} catch (err) {
// Non-fatal: default-on resolution is best-effort
storeLog.warn("Failed to auto-apply default workflow steps during task creation; auto-defaulting skipped", {
phase: "createTask:workflow-auto-default",
skippedAutoDefaulting: true,
@@ -2386,22 +2395,25 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
});
}
} else if (input.enabledWorkflowSteps.length === 0) {
// Explicitly empty array — user intentionally selected no steps
resolvedWorkflowSteps = undefined;
}
// Create the task immediately with current title (may be undefined)
const task = await this._createTaskInternal(input, title, resolvedWorkflowSteps, id);
const task = await this.createTaskWithDistributedReservation(input, {
createTaskWithId: async (taskId) => {
if (input.dependencies?.includes(taskId)) {
throw new Error(`Task ${taskId} cannot depend on itself`);
}
return this._createTaskInternal(input, title, resolvedWorkflowSteps, taskId);
},
});
// Fire async background handler for title summarization (non-blocking)
if (shouldSummarize && options?.onSummarize) {
const id = task.id;
Promise.resolve().then(async () => {
try {
const generatedTitle = await options.onSummarize!(input.description);
const normalizedTitle = sanitizeTitle(generatedTitle);
if (normalizedTitle) {
// Guard against races: read directly from SQLite to avoid extra
// prompt/step file I/O in this background path.
const currentTask = this.readTaskFromDb(id);
if (currentTask && !currentTask.title) {
await this.updateTask(id, { title: normalizedTitle });
@@ -2601,50 +2613,42 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
* execution state. The new task will be re-specified by the AI.
*/
async duplicateTask(id: string): Promise<Task> {
// Read the source task with its prompt
const sourceTask = await this.getTask(id);
// Allocate a new ID
const newId = await this.allocateId();
const now = new Date().toISOString();
// Create new task with copied title/description, but fresh state
const newTask: Task = {
id: newId,
lineageId: generateTaskLineageId(),
title: sourceTask.title,
description: `${sourceTask.description}\n\n(Duplicated from ${id})`,
priority: normalizeTaskPriority(sourceTask.priority),
column: "triage",
modelPresetId: sourceTask.modelPresetId,
sourceType: "task_duplicate",
sourceParentTaskId: id,
dependencies: [], // Fresh task should have no dependencies
steps: [], // Reset execution state
currentStep: 0,
log: [{ timestamp: now, action: `Duplicated from ${id}` }],
columnMovedAt: now,
createdAt: now,
updatedAt: now,
// Explicitly NOT copied: worktree, status, blockedBy, paused, executionStartBranch,
// attachments, comments, prInfo, agent logs, size, reviewLevel
baseBranch: sourceTask.baseBranch,
};
return this.createTaskWithDistributedReservation({ description: sourceTask.description }, {
createTaskWithId: async (newId) => {
const newTask: Task = {
id: newId,
lineageId: generateTaskLineageId(),
title: sourceTask.title,
description: `${sourceTask.description}\n\n(Duplicated from ${id})`,
priority: normalizeTaskPriority(sourceTask.priority),
column: "triage",
modelPresetId: sourceTask.modelPresetId,
sourceType: "task_duplicate",
sourceParentTaskId: id,
dependencies: [],
steps: [],
currentStep: 0,
log: [{ timestamp: now, action: `Duplicated from ${id}` }],
columnMovedAt: now,
createdAt: now,
updatedAt: now,
baseBranch: sourceTask.baseBranch,
};
const newDir = this.taskDir(newId);
await mkdir(newDir, { recursive: true });
await this.atomicWriteTaskJson(newDir, newTask);
const newDir = this.taskDir(newId);
await mkdir(newDir, { recursive: true });
await this.atomicWriteTaskJson(newDir, newTask);
await mkdir(newDir, { recursive: true });
await writeFile(join(newDir, "PROMPT.md"), sourceTask.prompt);
// Copy source PROMPT.md content (the AI will re-specify it in triage)
const sourcePrompt = sourceTask.prompt;
await mkdir(newDir, { recursive: true });
await writeFile(join(newDir, "PROMPT.md"), sourcePrompt);
// Update cache if watcher is active
if (this.isWatching) this.taskCache.set(newId, { ...newTask });
this.emit("task:created", newTask);
return newTask;
if (this.isWatching) this.taskCache.set(newId, { ...newTask });
this.emit("task:created", newTask);
return newTask;
},
});
}
/**
@@ -2653,27 +2657,19 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
* Validates the original is in 'done' or 'in-review' column.
*/
async refineTask(id: string, feedback: string): Promise<Task> {
// Read the source task with its prompt
const sourceTask = await this.getTask(id);
// Validate task is in done or in-review column
if (sourceTask.column !== "done" && sourceTask.column !== "in-review") {
throw new Error(
`Cannot refine ${id}: task is in '${sourceTask.column}', must be in 'done' or 'in-review'`,
);
}
// Validate feedback is not empty
if (!feedback?.trim()) {
throw new Error("Feedback is required and cannot be empty");
}
// Allocate a new ID
const newId = await this.allocateId();
const now = new Date().toISOString();
// Derive a readable source label for the refinement title.
// Precedence: title → first non-empty line of description (collapsed) → task ID
let sourceLabel: string;
if (sourceTask.title?.trim()) {
sourceLabel = sourceTask.title.trim();
@@ -2682,65 +2678,56 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
.split("\n")
.map((line: string) => line.trim())
.find((line: string) => line.length > 0);
if (firstLine) {
sourceLabel = firstLine.replace(/\s+/g, " ");
} else {
sourceLabel = sourceTask.id;
}
sourceLabel = firstLine ? firstLine.replace(/\s+/g, " ") : sourceTask.id;
}
// Create new refinement task
const newTask: Task = {
id: newId,
lineageId: generateTaskLineageId(),
title: `Refinement: ${sourceLabel}`,
description: `${feedback.trim()}\n\nRefines: ${id}`,
priority: normalizeTaskPriority(sourceTask.priority),
column: "triage",
dependencies: [id], // Refinement depends on the original being complete
sourceType: "task_refine",
sourceParentTaskId: id,
steps: [], // Reset execution state
currentStep: 0,
log: [{ timestamp: now, action: `Created as refinement of ${id}` }],
columnMovedAt: now,
createdAt: now,
updatedAt: now,
// Copy attachments from original for context (defensive copy)
attachments: sourceTask.attachments ? [...sourceTask.attachments] : undefined,
};
return this.createTaskWithDistributedReservation({ description: feedback.trim() }, {
createTaskWithId: async (newId) => {
const newTask: Task = {
id: newId,
lineageId: generateTaskLineageId(),
title: `Refinement: ${sourceLabel}`,
description: `${feedback.trim()}\n\nRefines: ${id}`,
priority: normalizeTaskPriority(sourceTask.priority),
column: "triage",
dependencies: [id],
sourceType: "task_refine",
sourceParentTaskId: id,
steps: [],
currentStep: 0,
log: [{ timestamp: now, action: `Created as refinement of ${id}` }],
columnMovedAt: now,
createdAt: now,
updatedAt: now,
attachments: sourceTask.attachments ? [...sourceTask.attachments] : undefined,
};
const newDir = this.taskDir(newId);
await mkdir(newDir, { recursive: true });
await this.atomicWriteTaskJson(newDir, newTask);
const newDir = this.taskDir(newId);
await mkdir(newDir, { recursive: true });
await this.atomicWriteTaskJson(newDir, newTask);
const prompt = `# ${newTask.title}\n\n${newTask.description}\n`;
await mkdir(newDir, { recursive: true });
await writeFile(join(newDir, "PROMPT.md"), prompt);
// Create a PROMPT.md for the refinement
const heading = newTask.title;
const prompt = `# ${heading}\n\n${newTask.description}\n`;
await mkdir(newDir, { recursive: true });
await writeFile(join(newDir, "PROMPT.md"), prompt);
// Copy attachments from source if any
if (sourceTask.attachments && sourceTask.attachments.length > 0) {
const sourceAttachDir = join(this.taskDir(id), "attachments");
const targetAttachDir = join(newDir, "attachments");
await mkdir(targetAttachDir, { recursive: true });
for (const attachment of sourceTask.attachments) {
const sourcePath = join(sourceAttachDir, attachment.filename);
const targetPath = join(targetAttachDir, attachment.filename);
if (existsSync(sourcePath)) {
const content = await readFile(sourcePath);
await writeFile(targetPath, content);
if (sourceTask.attachments && sourceTask.attachments.length > 0) {
const sourceAttachDir = join(this.taskDir(id), "attachments");
const targetAttachDir = join(newDir, "attachments");
await mkdir(targetAttachDir, { recursive: true });
for (const attachment of sourceTask.attachments) {
const sourcePath = join(sourceAttachDir, attachment.filename);
const targetPath = join(targetAttachDir, attachment.filename);
if (existsSync(sourcePath)) {
const content = await readFile(sourcePath);
await writeFile(targetPath, content);
}
}
}
}
}
// Update cache if watcher is active
if (this.isWatching) this.taskCache.set(newId, { ...newTask });
this.emit("task:created", newTask);
return newTask;
if (this.isWatching) this.taskCache.set(newId, { ...newTask });
this.emit("task:created", newTask);
return newTask;
},
});
}
/**

115
packages/dashboard/src/__tests__/routes-tasks.test.ts
View File

@@ -717,8 +717,8 @@ describe("POST /tasks", () => {
createIssueSpy.mockRestore();
});
it("uses distributed allocator flow when reserved-id create is available", async () => {
const createTaskWithReservedId = vi.fn().mockResolvedValue({
it("uses store.createTask for local task creation", async () => {
const createTask = vi.fn().mockResolvedValue({
...FAKE_TASK_DETAIL,
id: "FN-7001",
column: "triage",
@@ -726,14 +726,11 @@ describe("POST /tasks", () => {
updatedAt: "2026-05-05T00:00:00.000Z",
nodeId: "node-target",
});
const storeWithReservedCreate = createMockStore({
createTaskWithReservedId,
getTask: vi.fn().mockResolvedValue({ ...FAKE_TASK_DETAIL, prompt: "# FN-7001\n\nBig initiative\n" }),
});
const storeWithCreate = createMockStore({ createTask });
const app = express();
app.use(express.json());
app.use("/api", createApiRoutes(storeWithReservedCreate));
app.use("/api", createApiRoutes(storeWithCreate));
const res = await REQUEST(
app,
@@ -744,11 +741,11 @@ describe("POST /tasks", () => {
);
expect(res.status).toBe(201);
expect(createTaskWithReservedId).toHaveBeenCalledWith(
expect(createTask).toHaveBeenCalledWith(
expect.objectContaining({ description: "Big initiative", nodeId: "node-target" }),
expect.objectContaining({ taskId: "FN-7001" }),
expect.objectContaining({ settings: { autoSummarizeTitles: undefined } }),
);
expect((storeWithReservedCreate.getDistributedTaskIdAllocator as ReturnType<typeof vi.fn>).mock.results[0]?.value.commitDistributedTaskIdReservation).toHaveBeenCalled();
expect((storeWithCreate.getDistributedTaskIdAllocator as ReturnType<typeof vi.fn>)).not.toHaveBeenCalled();
});
it("returns 400 when nodeId is not a string", async () => {
@@ -764,39 +761,13 @@ describe("POST /tasks", () => {
expect(res.body.error).toContain("nodeId must be a string");
});
it("retries reserved-id create when first reservation overlaps an existing task id", async () => {
const reserveDistributedTaskId = vi
.fn()
.mockResolvedValueOnce({ reservationId: "res-1", taskId: "FN-7001" })
.mockResolvedValueOnce({ reservationId: "res-2", taskId: "FN-7002" });
const commitDistributedTaskIdReservation = vi.fn().mockResolvedValue({});
const abortDistributedTaskIdReservation = vi.fn().mockResolvedValue({});
const createTaskWithReservedId = vi
.fn()
.mockRejectedValueOnce(new Error("Task ID already exists: FN-7001"))
.mockResolvedValueOnce({
...FAKE_TASK_DETAIL,
id: "FN-7002",
column: "triage",
createdAt: "2026-05-05T00:00:00.000Z",
updatedAt: "2026-05-05T00:00:00.000Z",
});
const deleteTask = vi.fn().mockResolvedValue(undefined);
const getTask = vi.fn().mockResolvedValue({ ...FAKE_TASK_DETAIL, prompt: "# FN-7002\n\nBig initiative\n" });
const storeWithReservedCreate = createMockStore({
createTaskWithReservedId,
deleteTask,
getTask,
getDistributedTaskIdAllocator: vi.fn().mockReturnValue({
reserveDistributedTaskId,
commitDistributedTaskIdReservation,
abortDistributedTaskIdReservation,
}),
});
it("returns 500 when store.createTask throws", async () => {
const createTask = vi.fn().mockRejectedValue(new Error("Task ID already exists: FN-7001"));
const storeWithCreate = createMockStore({ createTask });
const app = express();
app.use(express.json());
app.use("/api", createApiRoutes(storeWithReservedCreate));
app.use("/api", createApiRoutes(storeWithCreate));
const res = await REQUEST(
app,
@@ -806,68 +777,8 @@ describe("POST /tasks", () => {
{ "Content-Type": "application/json" },
);
expect(res.status).toBe(201);
expect(createTaskWithReservedId).toHaveBeenCalledTimes(2);
expect(createTaskWithReservedId).toHaveBeenNthCalledWith(
1,
expect.objectContaining({ description: "Big initiative" }),
expect.objectContaining({ taskId: "FN-7001" }),
);
expect(createTaskWithReservedId).toHaveBeenNthCalledWith(
2,
expect.objectContaining({ description: "Big initiative" }),
expect.objectContaining({ taskId: "FN-7002" }),
);
expect(abortDistributedTaskIdReservation).toHaveBeenCalledTimes(1);
expect(abortDistributedTaskIdReservation).toHaveBeenCalledWith(expect.objectContaining({ reservationId: "res-1", reason: "failed-create" }));
expect(commitDistributedTaskIdReservation).toHaveBeenCalledWith(expect.objectContaining({ reservationId: "res-2" }));
expect(deleteTask).not.toHaveBeenCalled();
});
it("aborts reservation and deletes local task on replication failure", async () => {
const reserveDistributedTaskId = vi.fn().mockResolvedValue({ reservationId: "res-1", taskId: "FN-7002" });
const commitDistributedTaskIdReservation = vi.fn().mockResolvedValue({});
const abortDistributedTaskIdReservation = vi.fn().mockResolvedValue({});
const createTaskWithReservedId = vi.fn().mockResolvedValue({
...FAKE_TASK_DETAIL,
id: "FN-7002",
column: "triage",
createdAt: "2026-05-05T00:00:00.000Z",
updatedAt: "2026-05-05T00:00:00.000Z",
});
const deleteTask = vi.fn().mockResolvedValue(undefined);
const storeWithReservedCreate = createMockStore({
createTaskWithReservedId,
deleteTask,
getTask: vi.fn().mockResolvedValue({ ...FAKE_TASK_DETAIL, prompt: "# FN-7002\n\nBig initiative\n" }),
getDistributedTaskIdAllocator: vi.fn().mockReturnValue({
reserveDistributedTaskId,
commitDistributedTaskIdReservation,
abortDistributedTaskIdReservation,
}),
});
mockCentralListNodes.mockResolvedValue([{ id: "node-remote", type: "remote", url: "https://remote.example.com", apiKey: "secret" }]);
vi.stubGlobal("fetch", vi.fn().mockRejectedValue(new Error("network down")));
const app = express();
app.use(express.json());
app.use("/api", createApiRoutes(storeWithReservedCreate));
const res = await REQUEST(
app,
"POST",
"/api/tasks",
JSON.stringify({ description: "Big initiative" }),
{ "Content-Type": "application/json" },
);
expect(res.status).toBe(503);
expect(abortDistributedTaskIdReservation).toHaveBeenCalledWith(expect.objectContaining({ reservationId: "res-1", reason: "failed-create" }));
expect(deleteTask).toHaveBeenCalledWith("FN-7002");
expect(commitDistributedTaskIdReservation).not.toHaveBeenCalled();
expect(reserveDistributedTaskId).toHaveBeenCalledTimes(1);
vi.unstubAllGlobals();
mockCentralListNodes.mockResolvedValue([]);
expect(res.status).toBe(500);
expect(res.body.error).toContain("Task ID already exists: FN-7001");
});
it("forwards branch and baseBranch on create", async () => {

95
packages/dashboard/src/routes/register-task-workflow-routes.ts
View File

@@ -4,11 +4,9 @@ import {
COLUMNS,
TASK_PRIORITIES,
VALID_TRANSITIONS,
buildMeshReplicatedTaskCreatePayload,
isTaskPriority,
REPO_OVERRIDE_RE,
resolveTitleSummarizerSettingsModel,
toReplicatedCreateInput,
validateNodeOverrideChange,
canAgentTakeImplementationTaskForExplicitRouting,
formatRoleMismatchReason,
@@ -19,7 +17,6 @@ import { maybeCreateTrackingIssue } from "../github-tracking.js";
import { parseGitHubBadgeUrl } from "./register-git-github.js";
import { planTaskWorktreePath } from "@fusion/engine";
import { ApiError, badRequest, conflict, notFound } from "../api-error.js";
import { fetchFromRemoteNode } from "./register-settings-sync-helpers.js";
import type { ApiRoutesContext } from "./types.js";
import { resolveBranchSelection } from "./branch-selection.js";
@@ -350,91 +347,13 @@ export function registerTaskWorkflowRoutes(ctx: ApiRoutesContext, deps: TaskWork
...(validatedGithubTracking ? { githubTracking: validatedGithubTracking } : {}),
};
if (typeof scopedStore.createTaskWithReservedId !== "function") {
const task = await scopedStore.createTask(
createInput,
{ onSummarize, settings: { autoSummarizeTitles: settings.autoSummarizeTitles } },
);
await maybeCreateTaskTrackingIssue(scopedStore, task, options?.githubToken);
res.status(201).json(task);
return;
}
const allocator = scopedStore.getDistributedTaskIdAllocator();
const { CentralCore } = await import("@fusion/core");
const central = new CentralCore();
await central.init();
const nodes = await central.listNodes();
const localNode = nodes.find((node) => node.type === "local");
const remoteNodes = nodes.filter((node) => node.type === "remote" && node.url && node.apiKey);
await central.close();
const nodeIdForReservation = localNode?.id ?? "local";
const isOverlapClassFailure = (err: unknown): boolean => {
const message = err instanceof Error ? err.message : String(err);
return (
message.includes("Task ID already exists:") ||
message.includes("Replicated task payload collision for existing task")
);
};
const maxCreateAttempts = 3;
for (let attempt = 1; attempt <= maxCreateAttempts; attempt += 1) {
const reservation = await allocator.reserveDistributedTaskId({
prefix: "FN",
nodeId: nodeIdForReservation,
});
let createdTask: Task | null = null;
try {
createdTask = await scopedStore.createTaskWithReservedId(createInput, {
taskId: reservation.taskId,
});
await maybeCreateTaskTrackingIssue(scopedStore, createdTask, options?.githubToken);
const replicatedPayload = buildMeshReplicatedTaskCreatePayload({
taskId: createdTask.id,
reservationId: reservation.reservationId,
sourceNodeId: nodeIdForReservation,
createdAt: createdTask.createdAt,
updatedAt: createdTask.updatedAt,
prompt: (await scopedStore.getTask(createdTask.id)).prompt,
createInput: toReplicatedCreateInput(createdTask),
});
for (const peer of remoteNodes) {
await fetchFromRemoteNode(peer, "/api/mesh/tasks/create", {
method: "POST",
body: replicatedPayload,
});
}
await allocator.commitDistributedTaskIdReservation({
reservationId: reservation.reservationId,
nodeId: nodeIdForReservation,
});
res.status(201).json(createdTask);
return;
} catch (err: unknown) {
await allocator.abortDistributedTaskIdReservation({
reservationId: reservation.reservationId,
nodeId: nodeIdForReservation,
reason: "failed-create",
}).catch(() => undefined);
if (createdTask) {
await scopedStore.deleteTask(createdTask.id).catch(() => undefined);
}
if (attempt < maxCreateAttempts && isOverlapClassFailure(err)) {
continue;
}
const message = err instanceof Error ? err.message : String(err);
throw new ApiError(503, `Cluster task create failed: ${message}`);
}
}
const task = await scopedStore.createTask(
createInput,
{ onSummarize, settings: { autoSummarizeTitles: settings.autoSummarizeTitles } },
);
await maybeCreateTaskTrackingIssue(scopedStore, task, options?.githubToken);
res.status(201).json(task);
return;
} catch (err: unknown) {
if (err instanceof ApiError) {
throw err;