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

feat(FN-3450): add mesh task create replication across nodes

Browse Source 
Implements distributed mesh task creation by adding replicated create primitives to the core store, wiring new API routes (`register-mesh-routes.ts`) that replicate task creation across clustered nodes while preserving remote-targeting metadata, and updating the dashboard's task creation flow accord

Fusion-Task-Id: FN-3450
This commit is contained in:
Fusion
2026-05-07 00:05:34 -07:00
committed by gsxdsm
parent 967896fb16
commit 935166dbb5
18 changed files with 887 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
packages/dashboard/src/routes/register-mesh-routes.ts
View File

@@ -234,6 +234,35 @@ export const registerMeshRoutes: ApiRouteRegistrar = (ctx) => {
}
});
router.post("/mesh/tasks/create", async (req, res) => {
const payload = req.body;
try {
const senderNodeId = typeof payload?.sourceNodeId === "string" ? payload.sourceNodeId : undefined;
if (!(await requireMeshAuth(req, res, senderNodeId))) return;
if (payload?.replicationVersion !== 1) throw badRequest("replicationVersion must be 1");
if (typeof payload?.reservationId !== "string" || payload.reservationId.trim().length === 0) throw badRequest("reservationId is required");
if (typeof payload?.taskId !== "string" || payload.taskId.trim().length === 0) throw badRequest("taskId is required");
if (typeof payload?.sourceNodeId !== "string" || payload.sourceNodeId.trim().length === 0) throw badRequest("sourceNodeId is required");
if (typeof payload?.createdAt !== "string" || typeof payload?.updatedAt !== "string") throw badRequest("createdAt and updatedAt are required");
if (typeof payload?.prompt !== "string") throw badRequest("prompt is required");
if (!payload?.input || typeof payload.input !== "object") throw badRequest("input is required");
const result = await store.applyReplicatedTaskCreate(payload);
res.status(result.applied ? 201 : 200).json(result);
} catch (err: unknown) {
emitRemoteRouteDiagnostic({
route: "mesh-task-create",
message: "Failed to apply replicated task create",
nodeId: typeof payload?.sourceNodeId === "string" ? payload.sourceNodeId : undefined,
upstreamPath: "/api/mesh/tasks/create",
operationStage: "apply-replicated-create",
error: err,
});
if (err instanceof ApiError) throw err;
rethrowAsApiError(err);
}
});
router.post("/mesh/sync", async (req, res) => {
try {
const { CentralCore } = await import("@fusion/core");

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

@@ -4,12 +4,15 @@ import {
COLUMNS,
TASK_PRIORITIES,
VALID_TRANSITIONS,
buildMeshReplicatedTaskCreatePayload,
isTaskPriority,
resolveTitleSummarizerSettingsModel,
toReplicatedCreateInput,
validateNodeOverrideChange,
} from "@fusion/core";
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";
interface TaskWorkflowRouteDeps {
@@ -97,6 +100,7 @@ export function registerTaskWorkflowRoutes(ctx: ApiRoutesContext, deps: TaskWork
source,
branch,
baseBranch,
nodeId,
} = req.body;
if (!description || typeof description !== "string") {
throw badRequest("description is required");
@@ -136,6 +140,10 @@ export function registerTaskWorkflowRoutes(ctx: ApiRoutesContext, deps: TaskWork
throw badRequest(`priority must be one of: ${TASK_PRIORITIES.join(", ")}`);
}
if (nodeId !== undefined && nodeId !== null && typeof nodeId !== "string") {
throw badRequest("nodeId must be a string");
}
const executorModel = normalizeModelSelectionPair(validatedModelProvider, validatedModelId);
const validatorModel = normalizeModelSelectionPair(validatedValidatorModelProvider, validatedValidatorModelId);
const planningModel = normalizeModelSelectionPair(validatedPlanningModelProvider, validatedPlanningModelId);
@@ -198,33 +206,97 @@ export function registerTaskWorkflowRoutes(ctx: ApiRoutesContext, deps: TaskWork
const normalizedBranch = validateOptionalBranchString(branch, "branch");
const normalizedBaseBranch = validateOptionalBranchString(baseBranch, "baseBranch");
const task = await scopedStore.createTask(
{
title,
description,
column,
dependencies,
breakIntoSubtasks,
enabledWorkflowSteps,
modelPresetId: validateOptionalModelField(modelPresetId, "modelPresetId"),
modelProvider: executorModel.provider ?? undefined,
modelId: executorModel.modelId ?? undefined,
validatorModelProvider: validatorModel.provider ?? undefined,
validatorModelId: validatorModel.modelId ?? undefined,
planningModelProvider: planningModel.provider ?? undefined,
planningModelId: planningModel.modelId ?? undefined,
thinkingLevel: thinkingLevel || undefined,
summarize,
reviewLevel: reviewLevel ?? undefined,
executionMode: executionMode || undefined,
priority: priority ?? undefined,
source: normalizedSource,
branch: normalizedBranch,
baseBranch: normalizedBaseBranch,
},
{ onSummarize, settings: { autoSummarizeTitles: settings.autoSummarizeTitles } }
);
res.status(201).json(task);
const createInput = {
title,
description,
column,
dependencies,
breakIntoSubtasks,
enabledWorkflowSteps,
modelPresetId: validateOptionalModelField(modelPresetId, "modelPresetId"),
modelProvider: executorModel.provider ?? undefined,
modelId: executorModel.modelId ?? undefined,
validatorModelProvider: validatorModel.provider ?? undefined,
validatorModelId: validatorModel.modelId ?? undefined,
planningModelProvider: planningModel.provider ?? undefined,
planningModelId: planningModel.modelId ?? undefined,
thinkingLevel: thinkingLevel || undefined,
summarize,
reviewLevel: reviewLevel ?? undefined,
executionMode: executionMode || undefined,
priority: priority ?? undefined,
source: normalizedSource,
branch: normalizedBranch,
baseBranch: normalizedBaseBranch,
...(typeof nodeId === "string" && nodeId.trim().length > 0 ? { nodeId: nodeId.trim() } : {}),
};
if (typeof scopedStore.createTaskWithReservedId !== "function") {
const task = await scopedStore.createTask(
createInput,
{ onSummarize, settings: { autoSummarizeTitles: settings.autoSummarizeTitles } },
);
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 reservation = await allocator.reserveDistributedTaskId({
prefix: "FN",
nodeId: localNode?.id ?? "local",
});
let createdTask: Task | null = null;
try {
createdTask = await scopedStore.createTaskWithReservedId(createInput, {
taskId: reservation.taskId,
});
const replicatedPayload = buildMeshReplicatedTaskCreatePayload({
taskId: createdTask.id,
reservationId: reservation.reservationId,
sourceNodeId: localNode?.id ?? "local",
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: localNode?.id ?? "local",
});
res.status(201).json(createdTask);
} catch (err: unknown) {
await allocator.abortDistributedTaskIdReservation({
reservationId: reservation.reservationId,
nodeId: localNode?.id ?? "local",
reason: "failed-create",
}).catch(() => undefined);
if (createdTask) {
await scopedStore.deleteTask(createdTask.id).catch(() => undefined);
}
const message = err instanceof Error ? err.message : String(err);
throw new ApiError(503, `Cluster task create failed: ${message}`);
}
} catch (err: unknown) {
if (err instanceof ApiError) {
throw err;