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

fix(FN-1572): stabilize fusion agent execution

Browse Source
This commit is contained in:
gsxdsm
2026-04-17 12:13:39 -07:00
parent 9bcefcb3a9
commit f35ea101c1
29 changed files with 657 additions and 283 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -64,7 +64,35 @@ vi.mock("./worktree-names.js", async () => {
// promisify(exec) in executor.ts.
vi.mock("node:child_process", async () => {
const { promisify } = await import("node:util");
const { EventEmitter } = await import("node:events");
const execSyncFn = vi.fn();
const spawnFn = vi.fn((cmd: string, opts?: any) => {
const child = new EventEmitter() as any;
child.stdout = new EventEmitter();
child.stderr = new EventEmitter();
child.pid = 12345;
child.exitCode = null;
child.signalCode = null;
child.kill = vi.fn();
queueMicrotask(() => {
try {
const out = execSyncFn(cmd, opts);
const stdout = out === undefined ? "" : out.toString();
if (stdout) child.stdout.emit("data", Buffer.from(stdout));
child.exitCode = 0;
child.emit("close", 0, null);
} catch (err) {
const error = err as { stdout?: string; stderr?: string; status?: number; code?: number };
const stdout = error?.stdout?.toString?.() ?? "";
const stderr = error?.stderr?.toString?.() ?? "";
if (stdout) child.stdout.emit("data", Buffer.from(stdout));
if (stderr) child.stderr.emit("data", Buffer.from(stderr));
child.exitCode = error.status ?? error.code ?? 1;
child.emit("close", child.exitCode, null);
}
});
return child;
});
const execFn: any = vi.fn((cmd: string, opts: any, cb: any) => {
const callback = typeof opts === "function" ? opts : cb;
@@ -95,7 +123,7 @@ vi.mock("node:child_process", async () => {
}
});
});
return { execSync: execSyncFn, exec: execFn };
return { execSync: execSyncFn, exec: execFn, spawn: spawnFn };
});
vi.mock("node:fs", () => ({
existsSync: vi.fn().mockReturnValue(true),

53
packages/engine/src/executor.ts
View File

@@ -6,7 +6,7 @@ import { isAbsolute, join } from "node:path";
import { existsSync } from "node:fs";
import { readFile, writeFile } from "node:fs/promises";
import type { TaskStore, Task, TaskDetail, StepStatus, Settings, WorkflowStep, MissionStore, Slice, AgentState, AgentCapability, RunMutationContext } from "@fusion/core";
import { buildExecutionMemoryInstructions, getTaskMergeBlocker, resolveAgentPrompt } from "@fusion/core";
import { buildExecutionMemoryInstructions, getTaskMergeBlocker, resolveAgentPrompt, runCommandAsync, type RunCommandResult } from "@fusion/core";
import { findWorktreeUser } from "./merger.js";
import { generateWorktreeName, slugify } from "./worktree-names.js";
import { Type, type Static } from "@mariozechner/pi-ai";
@@ -78,6 +78,31 @@ function truncateWorkflowScriptOutput(output: string): string {
return `... output truncated to last ${WORKFLOW_SCRIPT_OUTPUT_MAX_CHARS} characters ...\n${output.slice(-WORKFLOW_SCRIPT_OUTPUT_MAX_CHARS)}`;
}
function configuredCommandErrorMessage(result: RunCommandResult): string {
if (result.spawnError) return result.spawnError.message;
const parts: string[] = [];
if (result.timedOut) parts.push("Timed out");
if (result.exitCode !== null) parts.push(`Exit code: ${result.exitCode}`);
if (result.signal) parts.push(`Signal: ${result.signal}`);
const stdout = result.stdout.trim();
const stderr = result.stderr.trim();
if (stdout) parts.push(`stdout: ${truncateWorkflowScriptOutput(stdout)}`);
if (stderr) parts.push(`stderr: ${truncateWorkflowScriptOutput(stderr)}`);
return parts.length ? parts.join("\n") : "Command failed";
}
async function runConfiguredCommand(
command: string,
cwd: string,
timeoutMs: number,
): Promise<RunCommandResult> {
return runCommandAsync(command, {
cwd,
timeoutMs,
maxBuffer: 10 * 1024 * 1024,
});
}
// ── Tool parameter schemas (module-level for reuse in ToolDefinition generics) ──
const taskUpdateParams = Type.Object({
@@ -1072,10 +1097,10 @@ export class TaskExecutor {
// while the user-configured command (e.g. `pnpm install`) executes.
if (settings.worktreeInitCommand) {
try {
await execAsync(settings.worktreeInitCommand, {
cwd: worktreePath,
timeout: 120_000,
});
const initResult = await runConfiguredCommand(settings.worktreeInitCommand, worktreePath, 120_000);
if (initResult.spawnError || initResult.timedOut || initResult.exitCode !== 0) {
throw new Error(configuredCommandErrorMessage(initResult));
}
await this.store.logEntry(task.id, "Worktree init command completed", settings.worktreeInitCommand, this.currentRunContext);
} catch (err: unknown) {
const execError = err instanceof Error ? err : new Error(String(err));
@@ -1091,10 +1116,10 @@ export class TaskExecutor {
const scriptCommand = settings.scripts?.[settings.setupScript];
if (scriptCommand) {
try {
await execAsync(scriptCommand, {
cwd: worktreePath,
timeout: 120_000,
});
const setupResult = await runConfiguredCommand(scriptCommand, worktreePath, 120_000);
if (setupResult.spawnError || setupResult.timedOut || setupResult.exitCode !== 0) {
throw new Error(configuredCommandErrorMessage(setupResult));
}
await this.store.logEntry(task.id, `Setup script '${settings.setupScript}' completed`, scriptCommand, this.currentRunContext);
} catch (err: unknown) {
const execError = err instanceof Error ? err : new Error(String(err));
@@ -3126,12 +3151,10 @@ ${failureFeedback}
await this.store.logEntry(task.id, `Workflow step '${workflowStep.name}' executing script '${scriptName}': ${scriptCommand}`);
try {
// Non-blocking: async exec so the executor event loop keeps running
// while the user-configured workflow script executes.
await execAsync(scriptCommand, {
cwd: worktreePath,
timeout: 120_000,
});
const scriptResult = await runConfiguredCommand(scriptCommand, worktreePath, 120_000);
if (scriptResult.spawnError || scriptResult.timedOut || scriptResult.exitCode !== 0) {
return { success: false, error: configuredCommandErrorMessage(scriptResult) };
}
return { success: true, output: `Script '${scriptName}' completed successfully` };
} catch (err: unknown) {
const execError = err instanceof Error ? err : new Error(String(err));

30
packages/engine/src/merger.test.ts
View File

@@ -19,7 +19,35 @@ vi.mock("./pi.js", () => ({
// resolves/rejects based on the callback wired here.
vi.mock("node:child_process", async () => {
const { promisify } = await import("node:util");
const { EventEmitter } = await import("node:events");
const execSyncFn = vi.fn();
const spawnFn = vi.fn((cmd: string, opts?: any) => {
const child = new EventEmitter() as any;
child.stdout = new EventEmitter();
child.stderr = new EventEmitter();
child.pid = 12345;
child.exitCode = null;
child.signalCode = null;
child.kill = vi.fn();
queueMicrotask(() => {
try {
const out = execSyncFn(cmd, opts);
const stdout = out === undefined ? "" : out.toString();
if (stdout) child.stdout.emit("data", Buffer.from(stdout));
child.exitCode = 0;
child.emit("close", 0, null);
} catch (err) {
const error = err as { stdout?: string; stderr?: string; status?: number; code?: number };
const stdout = error?.stdout?.toString?.() ?? "";
const stderr = error?.stderr?.toString?.() ?? "";
if (stdout) child.stdout.emit("data", Buffer.from(stdout));
if (stderr) child.stderr.emit("data", Buffer.from(stderr));
child.exitCode = error.status ?? error.code ?? 1;
child.emit("close", child.exitCode, null);
}
});
return child;
});
const execFn: any = vi.fn((cmd: any, opts: any, cb: any) => {
const callback = typeof opts === "function" ? opts : cb;
try {
@@ -45,7 +73,7 @@ vi.mock("node:child_process", async () => {
}
});
});
return { execSync: execSyncFn, exec: execFn };
return { execSync: execSyncFn, exec: execFn, spawn: spawnFn };
});
vi.mock("node:fs", () => ({

40
packages/engine/src/merger.ts
View File

@@ -5,7 +5,7 @@ import { promisify } from "node:util";
const execAsync = promisify(exec);
import { existsSync } from "node:fs";
import { join } from "node:path";
import { getTaskMergeBlocker, type TaskStore, type MergeResult, type MergeDetails, type WorkflowStep, type WorkflowStepResult, type Settings, type AgentPromptsConfig } from "@fusion/core";
import { getTaskMergeBlocker, runCommandAsync, type TaskStore, type MergeResult, type MergeDetails, type WorkflowStep, type WorkflowStepResult, type Settings, type AgentPromptsConfig } from "@fusion/core";
import { resolveAgentPrompt } from "@fusion/core";
import { createKbAgent, describeModel, promptWithFallback, compactSessionContext } from "./pi.js";
import { buildSessionSkillContext } from "./session-skill-context.js";
@@ -570,20 +570,36 @@ async function runVerificationCommand(
};
try {
// Execute the command with timeout (non-blocking: uses async exec so the
// engine event loop keeps running while the child process executes)
const { stdout, stderr } = await execAsync(command, {
const commandResult = await runCommandAsync(command, {
cwd: rootDir,
encoding: "utf-8",
timeoutMs: 300_000,
maxBuffer: VERIFICATION_COMMAND_MAX_BUFFER,
timeout: 300_000, // 5 minute timeout for verification commands
});
result.stdout = stdout;
result.stderr = stderr;
result.exitCode = 0;
result.success = true;
mergerLog.log(`${taskId}: ${type} command succeeded`);
await store.logEntry(taskId, `[verification] ${type} command succeeded (exit 0)`);
result.stdout = commandResult.stdout;
result.stderr = commandResult.stderr;
result.exitCode = commandResult.exitCode;
result.success = !commandResult.spawnError
&& !commandResult.timedOut
&& commandResult.exitCode === 0;
if (result.success) {
const bufferNote = commandResult.bufferExceeded ? ", output exceeded buffer" : "";
mergerLog.log(`${taskId}: ${type} command succeeded`);
await store.logEntry(taskId, `[verification] ${type} command succeeded (exit 0${bufferNote})`);
return result;
}
const failureText = commandResult.spawnError?.message
|| (commandResult.timedOut ? "Command timed out" : "")
|| commandResult.stderr
|| commandResult.stdout
|| `Command exited with ${commandResult.exitCode ?? commandResult.signal ?? "unknown status"}`;
throw Object.assign(new Error(failureText), {
stdout: commandResult.stdout,
stderr: commandResult.stderr,
status: commandResult.exitCode,
code: commandResult.timedOut ? "ETIMEDOUT" : undefined,
});
} catch (error: any) {
result.stdout = error.stdout?.toString() || "";
result.stderr = error.stderr?.toString() || "";

11
packages/engine/src/peer-exchange-service.test.ts
View File

@@ -111,6 +111,17 @@ describe("PeerExchangeService", () => {
expect(service).toBeDefined();
});
it("should use default sync interval of 120 seconds", () => {
mockListNodes.mockResolvedValue([]);
const setIntervalSpy = vi.spyOn(globalThis, "setInterval");
const service = new PeerExchangeService(mockCentralCore);
service.start();
expect(setIntervalSpy).toHaveBeenCalledWith(expect.any(Function), 120_000);
service.stop();
});
it("should default settingsSyncEnabled to false", async () => {
const service = new PeerExchangeService(mockCentralCore);
setupSuccessfulSync();

4
packages/engine/src/peer-exchange-service.ts
View File

@@ -3,7 +3,7 @@ import type { NodeConfig, PeerSyncRequest, PeerSyncResponse } from "@fusion/core
import { peerExchangeLog } from "./logger.js";
export interface PeerExchangeServiceOptions {
/** Interval between peer sync cycles in milliseconds. Default: 60000 (1 minute) */
/** Interval between peer sync cycles in milliseconds. Default: 120000 (2 minutes) */
syncIntervalMs?: number;
/** When true, include settings and model auth data in peer sync exchanges. Default: false. */
settingsSyncEnabled?: boolean;
@@ -70,7 +70,7 @@ export class PeerExchangeService {
*/
constructor(centralCore: CentralCore, options: PeerExchangeServiceOptions = {}) {
this.centralCore = centralCore;
this.syncIntervalMs = options.syncIntervalMs ?? 60_000; // 1 minute default
this.syncIntervalMs = options.syncIntervalMs ?? 120_000; // 2 minute default
this.settingsSyncEnabled = options.settingsSyncEnabled ?? false;
this.settingsSyncThrottleMs = options.settingsSyncThrottleMs ?? 300_000; // 5 minutes default
this.globalSettings = options.globalSettings;

47
packages/engine/src/pi-create-kb-agent.test.ts
View File

@@ -555,6 +555,53 @@ describe("createKbAgent", () => {
}));
});
it("falls back during prompt when the primary model rejects temperature settings", async () => {
const primaryPrompt = vi.fn().mockRejectedValue(
new Error("400 invalid temperature: only 0.6 is allowed for this model"),
);
const fallbackPrompt = vi.fn().mockResolvedValue(undefined);
const primaryDispose = vi.fn();
createAgentSessionMock
.mockResolvedValueOnce({
session: {
prompt: primaryPrompt,
subscribe: vi.fn(),
dispose: primaryDispose,
setThinkingLevel: vi.fn(),
},
})
.mockResolvedValueOnce({
session: {
prompt: fallbackPrompt,
subscribe: vi.fn(),
dispose: vi.fn(),
setThinkingLevel: vi.fn(),
},
});
const { createKbAgent } = await import("./pi.js");
const { session } = await createKbAgent({
cwd: "/tmp",
systemPrompt: "test",
tools: "readonly",
defaultProvider: "kimi-coding",
defaultModelId: "kimi-k2.6-preview",
fallbackProvider: "zai",
fallbackModelId: "glm-5.1",
});
await (session as any).promptWithFallback("review this spec");
expect(primaryPrompt).toHaveBeenCalledWith("review this spec");
expect(primaryDispose).toHaveBeenCalled();
expect(fallbackPrompt).toHaveBeenCalledWith("review this spec");
expect(createAgentSessionMock).toHaveBeenNthCalledWith(2, expect.objectContaining({
model: { provider: "zai", id: "glm-5.1" },
}));
});
it("enables auto-compaction to prevent context-window overflow", async () => {
const { createKbAgent } = await import("./pi.js");

3
packages/engine/src/pi.ts
View File

@@ -381,7 +381,8 @@ function isRetryableModelSelectionError(message: string): boolean {
|| normalized.includes("overloaded")
|| normalized.includes("quota")
|| normalized.includes("capacity")
|| normalized.includes("temporarily unavailable");
|| normalized.includes("temporarily unavailable")
|| normalized.includes("invalid temperature");
}
interface PackageManagerSettingsView {

66
packages/engine/src/self-healing.test.ts
View File

@@ -1120,6 +1120,72 @@ describe("SelfHealingManager", () => {
managerWithRecovery.stop();
});
it("moves stale in-review tasks with incomplete steps back to todo for retry", async () => {
const managerWithRecovery = new SelfHealingManager(store, {
rootDir: "/tmp/test-project",
});
(store.getSettings as ReturnType<typeof vi.fn>).mockResolvedValue({
taskStuckTimeoutMs: 1_000,
});
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([
{
id: "FN-1572",
column: "in-review",
paused: false,
status: null,
error: null,
worktree: "/tmp/test-project/.worktrees/fn-1572",
updatedAt: new Date(Date.now() - 5_000).toISOString(),
steps: [
{ name: "Preflight", status: "done" },
{ name: "Testing & Verification", status: "in-progress" },
],
workflowStepResults: [],
mergeDetails: undefined,
log: [],
},
]);
const result = await managerWithRecovery.recoverStaleIncompleteReviewTasks();
expect(result).toBe(1);
expect(store.logEntry).toHaveBeenCalledWith(
"FN-1572",
expect.stringContaining("in-review task still had incomplete steps"),
);
expect(store.moveTask).toHaveBeenCalledWith("FN-1572", "todo");
managerWithRecovery.stop();
});
it("does not move fresh in-review tasks with incomplete steps", async () => {
const managerWithRecovery = new SelfHealingManager(store, {
rootDir: "/tmp/test-project",
});
(store.getSettings as ReturnType<typeof vi.fn>).mockResolvedValue({
taskStuckTimeoutMs: 60_000,
});
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([
{
id: "FN-1573",
column: "in-review",
paused: false,
status: null,
updatedAt: new Date().toISOString(),
steps: [{ name: "Testing", status: "in-progress" }],
workflowStepResults: [],
log: [],
},
]);
const result = await managerWithRecovery.recoverStaleIncompleteReviewTasks();
expect(result).toBe(0);
expect(store.moveTask).not.toHaveBeenCalled();
managerWithRecovery.stop();
});
it("moves merged in-review tasks to done and clears transient merge state", async () => {
const managerWithRecovery = new SelfHealingManager(store, {
rootDir: "/tmp/test-project",

55
packages/engine/src/self-healing.ts
View File

@@ -64,6 +64,7 @@ export interface SelfHealingOptions {
const APPROVED_TRIAGE_RECOVERY_GRACE_MS = 60_000;
const ORPHANED_EXECUTION_RECOVERY_GRACE_MS = 60_000;
const ACTIVE_MERGE_STATUSES = new Set(["merging", "merging-pr"]);
const NON_TERMINAL_STEP_STATUSES = new Set(["pending", "in-progress"]);
/**
* Longer grace period for tasks that still have a worktree on disk.
* This avoids racing with `executor.resumeOrphaned()` which runs on
@@ -140,6 +141,7 @@ export class SelfHealingManager {
async runStartupRecovery(): Promise<void> {
await this.recoverNoProgressNoTaskDoneFailures();
await this.recoverCompletedTasks();
await this.recoverStaleIncompleteReviewTasks();
await this.recoverInterruptedMergingTasks();
await this.recoverMisclassifiedFailures();
await this.recoverOrphanedExecutions();
@@ -456,6 +458,7 @@ export class SelfHealingManager {
this.checkpointWal();
await this.enforceWorktreeCap();
await this.recoverCompletedTasks();
await this.recoverStaleIncompleteReviewTasks();
await this.recoverInterruptedMergingTasks();
await this.recoverMergeableReviewTasks();
await this.recoverMergedReviewTasks();
@@ -632,6 +635,58 @@ export class SelfHealingManager {
}
}
/**
* Recover tasks that reached `in-review` while a task step was still marked
* pending/in-progress. These tasks are not tracked by StuckTaskDetector
* anymore because the executor session is gone, and they are not mergeable
* because `getTaskMergeBlocker()` correctly blocks incomplete steps.
*
* Moving them back to `todo` lets the normal scheduler/executor resume the
* incomplete step instead of leaving the task stranded in review.
*/
async recoverStaleIncompleteReviewTasks(): Promise<number> {
try {
const settings = await this.store.getSettings();
const timeoutMs = settings.taskStuckTimeoutMs;
if (!timeoutMs || timeoutMs <= 0) return 0;
const now = Date.now();
const tasks = await this.store.listTasks({ column: "in-review" });
const staleIncomplete = tasks.filter((task) =>
task.column === "in-review" &&
!task.paused &&
!task.status &&
task.steps.length > 0 &&
task.steps.some((step) => NON_TERMINAL_STEP_STATUSES.has(step.status)) &&
now - new Date(task.updatedAt).getTime() >= timeoutMs
);
if (staleIncomplete.length === 0) return 0;
log.warn(`Found ${staleIncomplete.length} stale in-review task(s) with incomplete steps`);
let recovered = 0;
for (const task of staleIncomplete) {
try {
await this.store.logEntry(
task.id,
"Auto-recovered: in-review task still had incomplete steps — moved back to todo for retry",
);
await this.store.moveTask(task.id, "todo");
log.log(`Recovered stale incomplete review task ${task.id}: moved back to todo`);
recovered++;
} catch (err: unknown) { const errorMessage = err instanceof Error ? err.message : String(err);
log.error(`Failed to recover stale incomplete review task ${task.id}: ${errorMessage}`);
}
}
return recovered;
} catch (err: unknown) { const errorMessage = err instanceof Error ? err.message : String(err);
log.error(`Stale incomplete review recovery failed: ${errorMessage}`);
return 0;
}
}
/**
* Recover stale `in-review` tasks left in a transient merge status.
*

12
packages/engine/src/triage.test.ts
View File

@@ -1386,7 +1386,7 @@ describe("taskCreate tool model inheritance", () => {
});
describe("bounded recovery retries for triage", () => {
it("marks triage failed when the agent exits without calling review_spec", async () => {
it("requeues triage with backoff when the agent exits without calling review_spec", async () => {
const task = {
id: "FN-202",
description: "Test triage task",
@@ -1424,14 +1424,14 @@ describe("taskCreate tool model inheritance", () => {
await processor.specifyTask(task);
expect(store.updateTask).toHaveBeenCalledWith("FN-202", expect.objectContaining({
status: "failed",
error: expect.stringContaining("review_spec was never called"),
recoveryRetryCount: null,
nextRecoveryAt: null,
status: null,
error: null,
recoveryRetryCount: 1,
nextRecoveryAt: expect.any(String),
}));
expect(store.logEntry).toHaveBeenCalledWith(
"FN-202",
expect.stringContaining("Specification failed: spec review not approved"),
expect.stringContaining("Spec review not approved (review_spec was never called)"),
);
});

32
packages/engine/src/triage.ts
View File

@@ -902,19 +902,41 @@ export class TriageProcessor {
}
// Post-session APPROVE gate: only advance to todo when the spec
// reviewer explicitly approved. Any other verdict (REVISE,
// RETHINK, UNAVAILABLE) or a missing review (null) keeps the task
// in triage so unreviewed / rejected specs never reach execution.
// reviewer explicitly approved. Any other verdict (REVISE,
// RETHINK, UNAVAILABLE) or a missing review (null) stays in triage
// and is retried with bounded backoff instead of immediately failing.
if (specReviewVerdictRef.current !== "APPROVE") {
const verdictDesc =
specReviewVerdictRef.current === null
? "review_spec was never called"
: `verdict was ${specReviewVerdictRef.current}`;
const decision = computeRecoveryDecision({
recoveryRetryCount: task.recoveryRetryCount,
nextRecoveryAt: task.nextRecoveryAt,
});
if (decision.shouldRetry) {
const attempt = decision.nextState.recoveryRetryCount;
const delay = formatDelay(decision.delayMs);
const retryMessage =
`Spec review not approved (${verdictDesc}) — retry ${attempt}/${MAX_RECOVERY_RETRIES} in ${delay}.`;
triageLog.warn(`${task.id} ${retryMessage}`);
await this.store.logEntry(task.id, retryMessage);
const restoreStatus = task.status === "needs-respecify" ? "needs-respecify" : null;
await this.store.updateTask(task.id, {
status: restoreStatus,
error: null,
recoveryRetryCount: decision.nextState.recoveryRetryCount,
nextRecoveryAt: decision.nextState.nextRecoveryAt,
});
return;
}
const failureMessage =
`Specification failed: spec review not approved (${verdictDesc}). ` +
`Specification failed after ${MAX_RECOVERY_RETRIES} unapproved spec reviews (${verdictDesc}). ` +
"Retry after adjusting the task prompt or model.";
triageLog.log(
`${task.id} spec review not approved (${verdictDesc}) — marking specification failed`,
`${task.id} spec review not approved (${verdictDesc}) — retry budget exhausted`,
);
await this.store.logEntry(
task.id,