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

fix(FN-000): harden project migration runtime

Browse Source
This commit is contained in:
gsxdsm
2026-04-02 17:55:25 -07:00
parent 05f2114743
commit c5913b951d
10 changed files with 305 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
packages/engine/src/executor.test.ts
View File

@@ -3474,14 +3474,14 @@ describe("TaskExecutor usage limit detection", () => {
expect(onError).toHaveBeenCalled();
});
it("does NOT trigger global pause for non-usage-limit errors", async () => {
it("does NOT trigger global pause for transient non-usage-limit errors", async () => {
const store = createMockStore();
const pauser = new UsageLimitPauser(store);
const onUsageLimitHitSpy = vi.spyOn(pauser, "onUsageLimitHit");
const onError = vi.fn();
mockedCreateHaiAgent.mockRejectedValue(new Error("connection refused"));
const onError = vi.fn();
const executor = new TaskExecutor(store, "/tmp/test", {
onError,
usageLimitPauser: pauser,
@@ -3501,8 +3501,13 @@ describe("TaskExecutor usage limit detection", () => {
});
expect(onUsageLimitHitSpy).not.toHaveBeenCalled();
// Task should still be marked as failed
expect(store.updateTask).toHaveBeenCalledWith("FN-001", { status: "failed", error: "connection refused" });
expect(store.logEntry).toHaveBeenCalledWith("FN-001", "Transient error (will retry): connection refused");
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "todo");
expect(store.updateTask).not.toHaveBeenCalledWith(
"FN-001",
expect.objectContaining({ status: "failed" }),
);
expect(onError).not.toHaveBeenCalled();
});
it("works without usageLimitPauser (backward compatible)", async () => {

6
packages/engine/src/rate-limit-retry.test.ts
View File

@@ -52,12 +52,16 @@ describe("withRateLimitRetry", () => {
onRetry,
});
// Attach the rejection handler before advancing timers so the rejection
// is never unhandled when the final retry throws during timer advancement.
const assertion = expect(promise).rejects.toThrow("rate_limit exceeded");
// Advance enough to cover all backoff delays
for (let i = 0; i < 10; i++) {
await vi.advanceTimersByTimeAsync(500);
}
await expect(promise).rejects.toThrow("rate_limit exceeded");
await assertion;
expect(fn).toHaveBeenCalledTimes(3); // initial + 2 retries
expect(onRetry).toHaveBeenCalledTimes(2);
});

68
packages/engine/src/runtimes/in-process-runtime.ts
View File

@@ -3,11 +3,13 @@ import type {
TaskStore,
Task,
CentralCore,
AgentStore,
} from "@fusion/core";
import { Scheduler } from "../scheduler.js";
import { TaskExecutor, type TaskExecutorOptions } from "../executor.js";
import { WorktreePool } from "../worktree-pool.js";
import { AgentSemaphore } from "../concurrency.js";
import { HeartbeatMonitor } from "../agent-heartbeat.js";
import type {
ProjectRuntime,
ProjectRuntimeConfig,
@@ -63,6 +65,10 @@ export class InProcessRuntime
private globalSemaphore?: AgentSemaphore;
private stuckTaskDetector?: StuckTaskDetector;
private usageLimitPauser?: UsageLimitPauser;
private agentStore?: AgentStore;
private heartbeatMonitor?: HeartbeatMonitor;
/** Maps task IDs to agent IDs for lifecycle tracking */
private taskAgentMap = new Map<string, string>();
private lastActivityAt: string = new Date().toISOString();
/**
@@ -153,17 +159,41 @@ export class InProcessRuntime
onStart: (task, worktreePath) => {
this.recordActivity();
runtimeLog.log(`Started executing task ${task.id} in ${worktreePath}`);
// Create agent in AgentStore for lifecycle tracking
if (this.agentStore) {
this.agentStore.createAgent({
name: `executor-${task.id}`,
role: "executor",
}).then(async (agent: { id: string }) => {
this.taskAgentMap.set(task.id, agent.id);
await this.agentStore!.assignTask(agent.id, task.id);
await this.agentStore!.updateAgentState(agent.id, "active");
}).catch((err: unknown) => {
runtimeLog.warn(`Failed to create agent for task ${task.id}:`, err);
});
}
},
onComplete: (task) => {
this.recordActivity();
runtimeLog.log(`Completed task ${task.id}`);
// Record task completion in CentralCore
this.recordTaskCompletion(task.id, true);
// Update agent state to terminated (completed)
const agentId = this.taskAgentMap.get(task.id);
if (agentId && this.agentStore) {
void this.agentStore.updateAgentState(agentId, "terminated").catch(() => {});
this.taskAgentMap.delete(task.id);
}
},
onError: (task, error) => {
this.recordActivity();
runtimeLog.error(`Task ${task.id} failed:`, error.message);
this.recordTaskCompletion(task.id, false);
// Update agent state to terminated (failed)
const agentId = this.taskAgentMap.get(task.id);
if (agentId && this.agentStore) {
void this.agentStore.updateAgentState(agentId, "terminated").catch(() => {});
this.taskAgentMap.delete(task.id);
}
},
};
@@ -173,13 +203,35 @@ export class InProcessRuntime
executorOptions
);
// 6. Set up event forwarding from TaskStore
// 6. Initialize AgentStore and HeartbeatMonitor
try {
const { AgentStore: AgentStoreClass } = await import("@fusion/core");
this.agentStore = new AgentStoreClass({ rootDir: this.taskStore.getRootDir() });
await this.agentStore.init();
this.heartbeatMonitor = new HeartbeatMonitor({
store: this.agentStore,
onMissed: (agentId) => {
runtimeLog.warn(`Agent ${agentId} missed heartbeat`);
},
onTerminated: (agentId) => {
runtimeLog.warn(`Agent ${agentId} terminated (unresponsive)`);
},
});
this.heartbeatMonitor.start();
runtimeLog.log(`AgentStore and HeartbeatMonitor initialized`);
} catch (agentErr) {
// Non-fatal — agent monitoring is optional
runtimeLog.warn(`AgentStore initialization failed (continuing without agent monitoring):`, agentErr);
}
// 7. Set up event forwarding from TaskStore
this.setupEventForwarding();
// 7. Resume orphaned in-progress tasks
// 8. Resume orphaned in-progress tasks
await this.executor.resumeOrphaned();
// 8. Start scheduler
// 9. Start scheduler
this.scheduler.start();
this.setStatus("active");
@@ -214,7 +266,13 @@ export class InProcessRuntime
runtimeLog.log(`Stopping InProcessRuntime for project ${this.config.projectId}`);
try {
// 1. Stop scheduler (prevents new task scheduling)
// 1. Stop heartbeat monitor
if (this.heartbeatMonitor) {
this.heartbeatMonitor.stop();
runtimeLog.log("HeartbeatMonitor stopped");
}
// 2. Stop scheduler (prevents new task scheduling)
if (this.scheduler) {
this.scheduler.stop();
runtimeLog.log("Scheduler stopped");

56
packages/engine/src/scheduler.test.ts
View File

@@ -147,10 +147,19 @@ describe("Scheduler", () => {
});
it("triggers scheduling immediately when task:created event fires", async () => {
const store = createMockStore({
listTasks: vi.fn().mockResolvedValue([
// Mock filesystem validation so schedule() can proceed
vi.mocked(existsSync).mockReturnValue(true);
vi.mocked(readFile).mockResolvedValue("# Task\nDo something");
// First call (from start()) returns empty todo, second call (from event) returns the new task
const listTasksMock = vi.fn()
.mockResolvedValueOnce([]) // Initial schedule from start() sees no tasks
.mockResolvedValue([
createMockTask({ id: "FN-001", column: "todo", dependencies: [] }),
]),
]);
const store = createMockStore({
listTasks: listTasksMock,
getSettings: vi.fn().mockResolvedValue({ maxConcurrent: 2, maxWorktrees: 4 }),
updateTask: vi.fn().mockResolvedValue(undefined),
moveTask: vi.fn().mockResolvedValue(undefined),
@@ -159,17 +168,23 @@ describe("Scheduler", () => {
const scheduler = new Scheduler(store);
scheduler.start();
// Wait for initial schedule pass to complete
await new Promise((r) => setTimeout(r, 10));
// Find and call the task:created handler
const onCalls = (store.on as any).mock.calls;
const createdHandler = onCalls.find((call: any) => call[0] === "task:created")?.[1];
expect(createdHandler).toBeDefined();
// Simulate task:created event
const newTask = createMockTask({ id: "FN-002", column: "todo" });
// Simulate task:created event — triggers schedule() which now sees FN-001
const newTask = createMockTask({ id: "FN-001", column: "todo" });
await createdHandler(newTask);
// Wait for async schedule to complete
await new Promise((r) => setTimeout(r, 10));
// Verify schedule() was called (moveTask should be called since task can start)
expect(store.moveTask).toHaveBeenCalledWith("FN-002", "in-progress");
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "in-progress");
});
it("registers task:moved event listener", () => {
@@ -180,11 +195,24 @@ describe("Scheduler", () => {
});
it("triggers scheduling immediately when task:moved to done event fires", async () => {
// Mock filesystem validation so schedule() can proceed
vi.mocked(existsSync).mockReturnValue(true);
vi.mocked(readFile).mockResolvedValue("# Task\nDo something");
// Initially return only FN-001 in-progress so start() doesn't schedule FN-002
const listTasksMock = vi.fn()
.mockResolvedValueOnce([
createMockTask({ id: "FN-001", column: "in-progress", dependencies: [] }),
createMockTask({ id: "FN-002", column: "todo", dependencies: ["FN-001"] }),
])
// After event fires, FN-001 is done so FN-002's deps are satisfied
.mockResolvedValue([
createMockTask({ id: "FN-001", column: "done", dependencies: [] }),
createMockTask({ id: "FN-002", column: "todo", dependencies: ["FN-001"] }),
]);
const store = createMockStore({
listTasks: vi.fn().mockResolvedValue([
createMockTask({ id: "FN-001", column: "done", dependencies: [] }), // Completed dep
createMockTask({ id: "FN-002", column: "todo", dependencies: ["FN-001"] }), // Waiting on FN-001
]),
listTasks: listTasksMock,
getSettings: vi.fn().mockResolvedValue({ maxConcurrent: 2, maxWorktrees: 4 }),
updateTask: vi.fn().mockResolvedValue(undefined),
moveTask: vi.fn().mockResolvedValue(undefined),
@@ -193,15 +221,21 @@ describe("Scheduler", () => {
const scheduler = new Scheduler(store);
scheduler.start();
// Wait for initial schedule pass to complete
await new Promise((r) => setTimeout(r, 10));
// Find and call the task:moved handler
const onCalls = (store.on as any).mock.calls;
const movedHandler = onCalls.find((call: any) => call[0] === "task:moved")?.[1];
expect(movedHandler).toBeDefined();
// Simulate task:moved to done event
const doneTask = createMockTask({ id: "FN-001", column: "todo" });
const doneTask = createMockTask({ id: "FN-001", column: "in-progress" });
await movedHandler({ task: doneTask, from: "in-progress", to: "done" });
// Wait for async schedule to complete
await new Promise((r) => setTimeout(r, 10));
// Verify schedule() was called - FN-002 should now be able to start
expect(store.moveTask).toHaveBeenCalledWith("FN-002", "in-progress");
});

55
packages/engine/src/scheduler.ts
View File

@@ -1,4 +1,4 @@
import { resolveDependencyOrder, type TaskStore, type Task, type MissionStore, type FeatureStatus } from "@fusion/core";
import { resolveDependencyOrder, type TaskStore, type Task, type MissionStore } from "@fusion/core";
import { existsSync } from "node:fs";
import { readFile } from "node:fs/promises";
import { join } from "node:path";
@@ -163,6 +163,11 @@ export class Scheduler {
void this.handleMissionTaskStart(task.id, task.sliceId);
}
// Mission progress tracking: when task with sliceId moves to done
if (task.sliceId && this.options.missionStore && to === "done") {
void this.handleMissionTaskCompletion(task.id, task.sliceId);
}
// Event-driven scheduling: when a dependency completes (task moves to "done"),
// trigger scheduling immediately so waiting tasks can start without waiting
// for the next poll interval (up to 15 seconds).
@@ -533,6 +538,13 @@ export class Scheduler {
return;
}
if (feature.sliceId !== sliceId) {
schedulerLog.warn(
`Task ${taskId} sliceId ${sliceId} does not match linked feature ${feature.id} sliceId ${feature.sliceId}; skipping mission start update`,
);
return;
}
// Only update if feature is still in "triaged" status
if (feature.status === "triaged") {
await missionStore.updateFeatureStatus(feature.id, "in-progress");
@@ -543,6 +555,47 @@ export class Scheduler {
}
}
/**
* Handle mission task completion.
* When a task moves to "done", update the linked feature status to "done".
* updateFeatureStatus cascades via recomputeSliceStatus — if all features
* in the slice are done the slice status becomes "complete" automatically.
* We then call onSliceComplete to trigger auto-advance to the next slice.
*/
private async handleMissionTaskCompletion(taskId: string, sliceId: string): Promise<void> {
if (!this.options.missionStore) return;
const missionStore = this.options.missionStore;
try {
const feature = missionStore.getFeatureByTaskId(taskId);
if (!feature) return;
if (feature.sliceId !== sliceId) {
schedulerLog.warn(
`Task ${taskId} sliceId ${sliceId} does not match linked feature ${feature.id} sliceId ${feature.sliceId}; skipping mission completion update`,
);
return;
}
const sliceIdBeforeUpdate = feature.sliceId;
if (feature.status !== "done") {
missionStore.updateFeatureStatus(feature.id, "done");
schedulerLog.log(`Feature ${feature.id} marked done (task ${taskId} completed)`);
}
// Check if the slice became complete after the feature update
const slice = missionStore.getSlice(sliceIdBeforeUpdate);
if (slice && slice.status === "complete") {
schedulerLog.log(`Slice ${slice.id} is complete — triggering auto-advance`);
await this.onSliceComplete(slice);
}
} catch (err) {
schedulerLog.error(`Error handling mission task completion for ${taskId}:`, err);
}
}
async onSliceComplete(slice: import("@fusion/core").Slice): Promise<void> {
if (!this.options.missionStore) return;