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

feat(FN-2088): merge fusion/fn-2088

Browse Source
This commit is contained in:
gsxdsm
2026-04-18 23:22:26 -07:00
parent 456c8ed6e6
commit c77a4a572e
11 changed files with 756 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -3239,6 +3239,129 @@ describe("TaskExecutor pause behavior", () => {
});
});
describe("session tracking failure diagnostics", () => {
it("logs warning when sessionFile update fails during retry", async () => {
const warnSpy = vi.spyOn(executorLog, "warn");
const store = createMockStore();
const retrySessionFilePath = "/tmp/sessions/retry-failed.jsonl";
store.updateTask.mockImplementation(async (_taskId: string, patch: Record<string, unknown>) => {
if (patch?.sessionFile === retrySessionFilePath) {
throw new Error("retry sessionFile write failed");
}
return {};
});
mockedCreateHaiAgent
.mockResolvedValueOnce({
session: {
prompt: vi.fn().mockResolvedValue(undefined),
dispose: vi.fn(),
},
sessionFile: "/tmp/sessions/initial.jsonl",
} as any)
.mockResolvedValueOnce({
session: {
prompt: vi.fn().mockResolvedValue(undefined),
dispose: vi.fn(),
},
sessionFile: retrySessionFilePath,
} as any);
const executor = new TaskExecutor(store, "/tmp/test");
await expect(executor.execute({
id: "FN-001",
title: "Retry session task",
description: "Session retry diagnostics",
column: "in-progress",
dependencies: [],
steps: [],
currentStep: 0,
log: [],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
})).resolves.toBeUndefined();
expect(warnSpy).toHaveBeenCalledWith(
expect.stringContaining("FN-001: failed to update sessionFile during retry"),
);
warnSpy.mockRestore();
});
it("logs warning when sessionFile clear fails on completion", async () => {
const warnSpy = vi.spyOn(executorLog, "warn");
const store = createMockStore();
store.updateTask.mockImplementation(async (_taskId: string, patch: Record<string, unknown>) => {
if (patch?.sessionFile === null) {
throw new Error("session clear failed");
}
return {};
});
mockedCreateHaiAgent.mockResolvedValue({
session: {
prompt: vi.fn().mockResolvedValue(undefined),
dispose: vi.fn(),
},
sessionFile: "/tmp/sessions/clear-test.jsonl",
} as any);
const executor = new TaskExecutor(store, "/tmp/test");
await expect(executor.execute({
id: "FN-001",
title: "Session clear task",
description: "Session clear diagnostics",
column: "in-progress",
dependencies: [],
steps: [],
currentStep: 0,
log: [],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
})).resolves.toBeUndefined();
expect(warnSpy).toHaveBeenCalledWith(
expect.stringContaining("FN-001: failed to clear sessionFile on completion"),
);
warnSpy.mockRestore();
});
it("logs warning when child agent deletion fails during cleanup", async () => {
vi.useFakeTimers();
const warnSpy = vi.spyOn(executorLog, "warn");
try {
const store = createMockStore();
const agentStore = {
updateAgentState: vi.fn().mockResolvedValue(undefined),
deleteAgent: vi.fn().mockRejectedValue(new Error("delete failed")),
};
const executor = new TaskExecutor(store, "/tmp/test", {
agentStore: agentStore as any,
});
(executor as any).childSessions.set("child-007", {
dispose: vi.fn(),
});
await (executor as any).terminateChildAgent("child-007");
await vi.advanceTimersByTimeAsync(5000);
await Promise.resolve();
expect(warnSpy).toHaveBeenCalledWith(
expect.stringContaining("Failed to delete spawned child agent child-007"),
);
} finally {
warnSpy.mockRestore();
vi.useRealTimers();
}
});
});
describe("TaskExecutor executor model hot-swap", () => {
const buildUpdatedTask = (overrides: Partial<Task> = {}): Task => ({
id: "FN-001",

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

@@ -988,13 +988,16 @@ export class TaskExecutor {
try {
const ratingSummary = await this.options.agentStore.getRatingSummary(agent.id);
return await resolveAgentInstructions(agent, this.rootDir, ratingSummary);
} catch {
} catch (err: unknown) {
const msg = err instanceof Error ? err.message : String(err);
executorLog.warn(`${agent.id}: failed to load rating summary for instruction resolution, falling back to default instructions: ${msg}`);
return await resolveAgentInstructions(agent, this.rootDir);
}
}
}
} catch {
// Graceful fallback — no instructions if lookup fails
} catch (err: unknown) {
const msg = err instanceof Error ? err.message : String(err);
executorLog.warn(`Failed to resolve instructions for role '${role}', continuing without custom instructions: ${msg}`);
}
return "";
}
@@ -1429,7 +1432,10 @@ export class TaskExecutor {
onRetry: (attempt, delayMs, error) => {
const delaySec = Math.round(delayMs / 1000);
executorLog.warn(`⏳ ${task.id} rate limited — retry ${attempt} in ${delaySec}s: ${error.message}`);
this.store.logEntry(task.id, `Rate limited — retry ${attempt} in ${delaySec}s`, undefined, this.currentRunContext).catch(() => {});
this.store.logEntry(task.id, `Rate limited — retry ${attempt} in ${delaySec}s`, undefined, this.currentRunContext).catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
executorLog.warn(`${task.id}: failed to log rate-limit entry during step-session execution: ${msg}`);
});
},
});
@@ -1471,8 +1477,9 @@ export class TaskExecutor {
await execAsync(`git worktree remove "${worktreePath}" --force`, { cwd: this.rootDir });
// Audit trail: record worktree removal (FN-1404)
await audit.git({ type: "worktree:remove", target: worktreePath });
} catch {
// Worktree removal failed - ignoring since we're cleaning up anyway
} catch (wtErr: unknown) {
const msg = wtErr instanceof Error ? wtErr.message : String(wtErr);
executorLog.warn(`${task.id}: worktree removal failed during transient-error retry cleanup (${worktreePath}): ${msg}`);
}
}
await this.store.updateTask(task.id, {
@@ -1527,8 +1534,9 @@ export class TaskExecutor {
if (worktreePath && existsSync(worktreePath)) {
try {
await execAsync(`git worktree remove "${worktreePath}" --force`, { cwd: this.rootDir });
} catch {
// Worktree removal failed - ignoring since we're cleaning up anyway
} catch (wtErr: unknown) {
const msg = wtErr instanceof Error ? wtErr.message : String(wtErr);
executorLog.warn(`${task.id}: worktree removal failed during stuck-requeue cleanup (${worktreePath}): ${msg}`);
}
}
await this.store.updateTask(task.id, { status: "stuck-killed", worktree: null, branch: null });
@@ -1907,7 +1915,10 @@ export class TaskExecutor {
});
// Update session file for the retry session (so pause/resume works)
if (retrySessionFile) {
this.store.updateTask(task.id, { sessionFile: retrySessionFile }).catch(() => {});
this.store.updateTask(task.id, { sessionFile: retrySessionFile }).catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
executorLog.warn(`${task.id}: failed to update sessionFile during retry: ${msg}`);
});
}
// Reassign so finally{} disposes the correct session
@@ -1993,7 +2004,10 @@ export class TaskExecutor {
// Check both the local flag (graceful exit) and the instance set
// (error path where dispose caused prompt to throw).
if (!wasPaused && !this.pausedAborted.has(task.id)) {
this.store.updateTask(task.id, { sessionFile: null }).catch(() => {});
this.store.updateTask(task.id, { sessionFile: null }).catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
executorLog.warn(`${task.id}: failed to clear sessionFile on completion: ${msg}`);
});
}
// Invoke plugin onAgentRunEnd hook (fire-and-forget)
// eslint-disable-next-line @typescript-eslint/no-explicit-any
@@ -2005,7 +2019,10 @@ export class TaskExecutor {
onRetry: (attempt, delayMs, error) => {
const delaySec = Math.round(delayMs / 1000);
executorLog.warn(`⏳ ${task.id} rate limited — retry ${attempt} in ${delaySec}s: ${error.message}`);
this.store.logEntry(task.id, `Rate limited — retry ${attempt} in ${delaySec}s`, undefined, this.currentRunContext).catch(() => {});
this.store.logEntry(task.id, `Rate limited — retry ${attempt} in ${delaySec}s`, undefined, this.currentRunContext).catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
executorLog.warn(`${task.id}: failed to log rate-limit entry during regular execution: ${msg}`);
});
},
});
@@ -2587,7 +2604,9 @@ export class TaskExecutor {
try {
await sessionRef.current.navigateTree(checkpointId, { summarize: false });
executorLog.log(`${taskId}: RETHINK — session rewound to checkpoint ${checkpointId}`);
} catch {
} catch (rewindErr: unknown) {
const msg = rewindErr instanceof Error ? rewindErr.message : String(rewindErr);
executorLog.warn(`${taskId}: RETHINK navigateTree rewind failed, falling back to branchWithSummary: ${msg}`);
// Fallback to branchWithSummary
try {
sessionRef.current.sessionManager.branchWithSummary(
@@ -2651,8 +2670,9 @@ export class TaskExecutor {
// Remove worktree
try {
await execAsync(`git worktree remove "${worktreePath}" --force`, { cwd: this.rootDir });
} catch {
// Worktree may already be gone
} catch (err: unknown) {
const msg = err instanceof Error ? err.message : String(err);
executorLog.warn(`${taskId}: failed to remove worktree during dep-abort cleanup (${worktreePath}): ${msg}`);
}
// Delete the branch — use stored branch name if available, fall back to convention
@@ -2660,8 +2680,9 @@ export class TaskExecutor {
const branch = task.branch || `fusion/${taskId.toLowerCase()}`;
try {
await execAsync(`git branch -D "${branch}"`, { cwd: this.rootDir });
} catch {
// Branch may not exist
} catch (err: unknown) {
const msg = err instanceof Error ? err.message : String(err);
executorLog.warn(`${taskId}: failed to delete branch during dep-abort cleanup (${branch}): ${msg}`);
}
// Clear worktree tracking
@@ -3033,7 +3054,9 @@ ${failureFeedback}
encoding: "utf-8",
});
baseRef = stdout.trim();
} catch {
} catch (mergeBaseErr: unknown) {
const mergeBaseMsg = mergeBaseErr instanceof Error ? mergeBaseErr.message : String(mergeBaseErr);
executorLog.warn(`Failed merge-base lookup for diff base in ${worktreePath}, trying HEAD~1 fallback: ${mergeBaseMsg}`);
// If merge-base fails, use HEAD~1 as last resort
try {
const { stdout } = await execAsync("git rev-parse HEAD~1", {
@@ -3775,8 +3798,9 @@ and show an appropriate message to the user.\`
cwd: this.rootDir,
});
await this.store.logEntry(taskId, `Unlocked worktree`, worktreePath);
} catch {
// Unlock failed - worktree wasn't locked, that's fine
} catch (err: unknown) {
const msg = err instanceof Error ? err.message : String(err);
executorLog.warn(`${taskId}: failed to unlock conflicting worktree ${worktreePath} before cleanup: ${msg}`);
}
// Remove the worktree
@@ -3791,8 +3815,9 @@ and show an appropriate message to the user.\`
cwd: this.rootDir,
});
await this.store.logEntry(taskId, `Deleted branch`, branch);
} catch {
// Branch might not exist, that's fine
} catch (err: unknown) {
const msg = err instanceof Error ? err.message : String(err);
executorLog.warn(`${taskId}: failed to delete conflicting branch ${branch}: ${msg}`);
}
return true;
@@ -3990,7 +4015,9 @@ and show an appropriate message to the user.\`
`Reset ${completedSteps.length} step(s) to pending — branch had no commits (uncommitted work lost with worktree)`,
);
}
} catch {
} catch (err: unknown) {
const msg = err instanceof Error ? err.message : String(err);
executorLog.warn(`${task.id}: step-reset-on-work-lost failed (non-fatal, steps keep current status): ${msg}`);
// Branch may not exist or git commands may fail — non-fatal.
// Steps keep their current status (safe default: agent can
// inspect the worktree and decide).
@@ -4150,14 +4177,18 @@ and show an appropriate message to the user.\`
try {
await this.options.agentStore?.updateAgentState(childId, "terminated");
} catch {
// Agent may not exist in store — that's ok for cleanup
} catch (err: unknown) {
const msg = err instanceof Error ? err.message : String(err);
executorLog.warn(`Failed to update spawned child ${childId} state to 'terminated' during cleanup: ${msg}`);
}
// Auto-delete the child agent after a short delay so the UI can observe
// the terminal state before the agent is removed.
void setTimeout(() => {
this.options.agentStore?.deleteAgent(childId).catch(() => {});
this.options.agentStore?.deleteAgent(childId).catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
executorLog.warn(`Failed to delete spawned child agent ${childId}: ${msg}`);
});
}, 5000);
this.totalSpawnedCount = Math.max(0, this.totalSpawnedCount - 1);
@@ -4174,8 +4205,9 @@ and show an appropriate message to the user.\`
): Promise<void> {
try {
await this.options.agentStore?.updateAgentState(agentId, "running");
} catch {
// State update failure shouldn't block execution
} catch (err: unknown) {
const msg = err instanceof Error ? err.message : String(err);
executorLog.warn(`Failed to update spawned child ${agentId} state to 'running': ${msg}`);
}
try {

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

@@ -54,6 +54,9 @@ export const executorLog = createLogger("executor");
/** Logger for the triage processor subsystem. */
export const triageLog = createLogger("triage");
/** Logger for the AI session (pi) subsystem. */
export const piLog = createLogger("pi");
/** Logger for the merge/auto-merge subsystem. */
export const mergerLog = createLogger("merger");

71
packages/engine/src/pi.test.ts
View File

@@ -1,6 +1,7 @@
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
import { describeModel, compactSessionContext, COMPACTION_FALLBACK_INSTRUCTIONS, createKbAgent, promptWithFallback, type AgentOptions } from "./pi.js";
import type { AgentSession } from "@mariozechner/pi-coding-agent";
import { createAgentSession, type AgentSession } from "@mariozechner/pi-coding-agent";
import { piLog } from "./logger.js";
// Mock skill resolver functions - define inside factory to avoid hoisting issues
vi.mock("./skill-resolver.js", () => {
@@ -538,3 +539,71 @@ describe("promptWithFallback auto-compaction", () => {
}
});
});
describe("session failure diagnostics", () => {
it("logs warning when compaction fails during promptWithFallback", async () => {
const warnSpy = vi.spyOn(piLog, "warn");
const session = {
prompt: vi.fn().mockRejectedValueOnce(
new Error("prompt is too long: 210000 tokens > 200000 maximum"),
),
compact: vi.fn().mockRejectedValue(new Error("compaction exploded")),
} as unknown as AgentSession;
await expect(promptWithFallback(session, "test prompt")).rejects.toThrow(
"prompt is too long: 210000 tokens > 200000 maximum",
);
expect(warnSpy).toHaveBeenCalledWith(
expect.stringContaining("Context compaction failed (will fall through to kill/requeue): compaction exploded"),
);
warnSpy.mockRestore();
});
it("logs warning when session dispose fails during model fallback swap", async () => {
const warnSpy = vi.spyOn(piLog, "warn");
const createAgentSessionMock = vi.mocked(createAgentSession);
const primarySession = {
model: { provider: "test", id: "primary-model" },
prompt: vi.fn().mockRejectedValue(new Error("429 Too Many Requests")),
dispose: vi.fn(() => {
throw new Error("dispose failed");
}),
subscribe: vi.fn(),
setThinkingLevel: vi.fn(),
sessionFile: undefined,
} as unknown as AgentSession;
const fallbackSession = {
model: { provider: "test", id: "fallback-model" },
prompt: vi.fn().mockResolvedValue(undefined),
dispose: vi.fn(),
subscribe: vi.fn(),
setThinkingLevel: vi.fn(),
sessionFile: undefined,
} as unknown as AgentSession;
createAgentSessionMock
.mockResolvedValueOnce({ session: primarySession } as any)
.mockResolvedValueOnce({ session: fallbackSession } as any);
const { session } = await createKbAgent({
cwd: "/test/project",
systemPrompt: "Test fallback swap",
defaultProvider: "test",
defaultModelId: "primary-model",
fallbackProvider: "test",
fallbackModelId: "fallback-model",
});
await expect((session as any).promptWithFallback("Run task")).resolves.toBeUndefined();
expect(warnSpy).toHaveBeenCalledWith(
expect.stringContaining("Failed to dispose session during model fallback swap: dispose failed"),
);
warnSpy.mockRestore();
});
});

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

@@ -34,6 +34,7 @@ import {
} from "./skill-resolver.js";
import { isContextLimitError } from "./context-limit-detector.js";
import { createFusionAuthStorage, getModelRegistryModelsPath } from "./auth-storage.js";
import { piLog } from "./logger.js";
export interface AgentResult {
session: AgentSession;
@@ -292,8 +293,9 @@ export async function compactSessionContext(
};
}
return null;
} catch {
// Compaction failed — return null so caller can fall through to kill/requeue
} catch (err: unknown) {
const msg = err instanceof Error ? err.message : String(err);
piLog.warn(`Context compaction failed (will fall through to kill/requeue): ${msg}`);
return null;
}
}
@@ -826,8 +828,9 @@ export async function createKbAgent(options: AgentOptions): Promise<AgentResult>
usingFallback = true;
try {
session.dispose();
} catch {
// ignore dispose errors while swapping sessions
} catch (err: unknown) {
const msg = err instanceof Error ? err.message : String(err);
piLog.warn(`Failed to dispose session during model fallback swap: ${msg}`);
}
const fallbackSessionResult = await createSessionWithModel(fallbackModel);

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

@@ -5,6 +5,7 @@ import { randomUUID } from "node:crypto";
import type { Task, TaskStore, CentralCore, AgentStore, Agent } from "@fusion/core";
import { InProcessRuntime } from "./in-process-runtime.js";
import type { ProjectRuntimeConfig } from "../project-runtime.js";
import { runtimeLog } from "../logger.js";
const {
mockSelfHealingStart,
@@ -642,6 +643,84 @@ describe("InProcessRuntime", () => {
}, 30000);
});
describe("agent cleanup failure diagnostics", () => {
it("logs warning when agent state update fails on task completion", async () => {
const warnSpy = vi.spyOn(runtimeLog, "warn");
await runtime.start();
const store = getAgentStore(runtime);
const updateStateSpy = vi.spyOn(store, "updateAgentState").mockImplementation(async (_agentId, state) => {
if (state === "terminated") {
throw new Error("state update failed");
}
return {} as Agent;
});
const executorOptions = mockExecutorCtor.mock.calls.at(-1)?.[0] as {
onStart?: (task: Task, worktreePath: string) => void;
onComplete?: (task: Task) => void;
};
executorOptions.onStart?.({ id: "FN-DIAG-1" } as Task, join(testDir, "worktree-FN-DIAG-1"));
await vi.waitFor(async () => {
const agents = await store.listAgents({ includeEphemeral: true });
expect(agents.some((a: Agent) => a.name === "executor-FN-DIAG-1")).toBe(true);
});
updateStateSpy.mockClear();
executorOptions.onComplete?.({ id: "FN-DIAG-1" } as Task);
await Promise.resolve();
expect(warnSpy).toHaveBeenCalledWith(
expect.stringContaining("Failed to update agent"),
);
expect(warnSpy).toHaveBeenCalledWith(
expect.stringContaining("terminated (completion)"),
);
warnSpy.mockRestore();
}, 30000);
it("logs warning when agent deletion fails after task error", async () => {
vi.useFakeTimers();
const warnSpy = vi.spyOn(runtimeLog, "warn");
try {
await runtime.start();
const store = getAgentStore(runtime);
const deleteAgentSpy = vi.spyOn(store, "deleteAgent").mockRejectedValue(new Error("delete failed"));
const executorOptions = mockExecutorCtor.mock.calls.at(-1)?.[0] as {
onStart?: (task: Task, worktreePath: string) => void;
onError?: (task: Task, error: Error) => void;
};
executorOptions.onStart?.({ id: "FN-DIAG-2" } as Task, join(testDir, "worktree-FN-DIAG-2"));
await vi.waitFor(async () => {
const agents = await store.listAgents({ includeEphemeral: true });
expect(agents.some((a: Agent) => a.name === "executor-FN-DIAG-2")).toBe(true);
});
deleteAgentSpy.mockClear();
executorOptions.onError?.({ id: "FN-DIAG-2" } as Task, new Error("Task failed"));
await vi.advanceTimersByTimeAsync(5000);
await Promise.resolve();
expect(warnSpy).toHaveBeenCalledWith(
expect.stringContaining("Failed to delete agent"),
);
expect(warnSpy).toHaveBeenCalledWith(
expect.stringContaining("after error"),
);
} finally {
warnSpy.mockRestore();
vi.useRealTimers();
}
}, 30000);
});
describe("configuration", () => {
it("should store projectId in config", () => {
// Access via the constructor params - runtime is created with testDir

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

@@ -355,12 +355,18 @@ export class InProcessRuntime
// Update agent state to terminated (completed)
const agentId = this.taskAgentMap.get(task.id);
if (agentId && this.agentStore) {
void this.agentStore.updateAgentState(agentId, "terminated").catch(() => {});
void this.agentStore.updateAgentState(agentId, "terminated").catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
runtimeLog.warn(`Failed to update agent ${agentId} state to terminated (completion): ${msg}`);
});
this.taskAgentMap.delete(task.id);
// Auto-delete the task-worker agent after a short delay so the UI
// can observe the terminal state before the agent is removed.
void setTimeout(() => {
this.agentStore?.deleteAgent(agentId).catch(() => {});
this.agentStore?.deleteAgent(agentId).catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
runtimeLog.warn(`Failed to delete agent ${agentId} after completion: ${msg}`);
});
}, 5000);
}
},
@@ -387,12 +393,18 @@ export class InProcessRuntime
// Update agent state to terminated (failed)
const agentId = this.taskAgentMap.get(task.id);
if (agentId && this.agentStore) {
void this.agentStore.updateAgentState(agentId, "terminated").catch(() => {});
void this.agentStore.updateAgentState(agentId, "terminated").catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
runtimeLog.warn(`Failed to update agent ${agentId} state to terminated (error): ${msg}`);
});
this.taskAgentMap.delete(task.id);
// Auto-delete the task-worker agent after a short delay so the UI
// can observe the terminal state before the agent is removed.
void setTimeout(() => {
this.agentStore?.deleteAgent(agentId).catch(() => {});
this.agentStore?.deleteAgent(agentId).catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
runtimeLog.warn(`Failed to delete agent ${agentId} after error: ${msg}`);
});
}, 5000);
}
},
@@ -1021,7 +1033,9 @@ export class InProcessRuntime
try {
const state = await this.centralCore.getGlobalConcurrencyState();
return state.globalMaxConcurrent;
} catch {
} catch (err: unknown) {
const msg = err instanceof Error ? err.message : String(err);
runtimeLog.warn(`Failed to fetch global concurrency from CentralCore, falling back to default (4): ${msg}`);
// Fallback to default if CentralCore is unavailable
return 4;
}

126
packages/engine/src/step-session-executor.test.ts
View File

@@ -553,19 +553,29 @@ vi.mock("./logger.js", () => {
warn: vi.fn(),
error: vi.fn(),
});
const loggers = new Map<string, ReturnType<typeof createMockLogger>>();
const getLogger = (prefix: string) => {
if (!loggers.has(prefix)) {
loggers.set(prefix, createMockLogger());
}
return loggers.get(prefix)!;
};
return {
createLogger: vi.fn(() => createMockLogger()),
schedulerLog: createMockLogger(),
executorLog: createMockLogger(),
triageLog: createMockLogger(),
mergerLog: createMockLogger(),
worktreePoolLog: createMockLogger(),
reviewerLog: createMockLogger(),
prMonitorLog: createMockLogger(),
runtimeLog: createMockLogger(),
ipcLog: createMockLogger(),
projectManagerLog: createMockLogger(),
autopilotLog: createMockLogger(),
createLogger: vi.fn((prefix: string) => getLogger(prefix)),
schedulerLog: getLogger("scheduler"),
executorLog: getLogger("executor"),
triageLog: getLogger("triage"),
mergerLog: getLogger("merger"),
worktreePoolLog: getLogger("worktree-pool"),
reviewerLog: getLogger("reviewer"),
prMonitorLog: getLogger("pr-monitor"),
runtimeLog: getLogger("runtime"),
stepExecLog: getLogger("step-session-executor"),
ipcLog: getLogger("ipc"),
projectManagerLog: getLogger("project-manager"),
autopilotLog: getLogger("autopilot"),
};
});
@@ -633,10 +643,18 @@ import { createKbAgent } from "./pi.js";
import { generateWorktreeName } from "./worktree-names.js";
import { execSync } from "node:child_process";
import { AgentSemaphore } from "./concurrency.js";
import { createLogger } from "./logger.js";
const mockedCreateKbAgent = vi.mocked(createKbAgent);
const mockedExecSync = vi.mocked(execSync);
const mockedGenerateWorktreeName = vi.mocked(generateWorktreeName);
const mockedCreateLogger = vi.mocked(createLogger);
const getStepSessionLogger = () => mockedCreateLogger("step-session-executor") as {
log: ReturnType<typeof vi.fn>;
warn: ReturnType<typeof vi.fn>;
error: ReturnType<typeof vi.fn>;
};
function makeMockSession(promptFn?: () => Promise<void>) {
return {
@@ -1064,6 +1082,54 @@ describe("StepSessionExecutor", () => {
expect(results.every((r) => r.success)).toBe(true);
});
it("logs warning when cherry-pick --abort fails", async () => {
const prompt = `# Task: FN-001
### Step 0: Core work
- \`packages/core/src/types.ts\`
### Step 1: Engine work
- \`packages/engine/src/pi.ts\``;
const task = makeTaskDetail({
prompt,
steps: [
{ name: "Step 0", status: "pending" },
{ name: "Step 1", status: "pending" },
],
});
const settings = makeSettings({ maxParallelSteps: 2 });
const session = makeMockSession();
mockedCreateKbAgent.mockResolvedValue({ session } as any);
mockedExecSync.mockImplementation((cmd: string) => {
if (typeof cmd === "string" && cmd.includes("git log")) {
return "abc123def Some commit";
}
if (typeof cmd === "string" && cmd.includes("git cherry-pick") && cmd.includes("--abort")) {
throw new Error("abort failed");
}
if (typeof cmd === "string" && cmd.includes("git cherry-pick") && !cmd.includes("--abort")) {
throw new Error("Merge conflict");
}
return "";
});
const executor = new StepSessionExecutor({
taskDetail: task,
worktreePath: "/project/.worktrees/main",
rootDir: "/project",
settings,
});
await executor.executeAll();
expect(getStepSessionLogger().warn).toHaveBeenCalledWith(
expect.stringContaining("Cherry-pick --abort failed for step"),
);
});
it("semaphore integration: parallel steps acquire/release", async () => {
const prompt = `# Task: FN-001
@@ -1488,6 +1554,42 @@ describe("StepSessionExecutor", () => {
});
});
describe("cleanup failure diagnostics", () => {
it("logs warning when session dispose fails during error cleanup", async () => {
const task = makeTaskDetail({
prompt: makeStepPrompt("FN-001", 1),
steps: [{ name: "Step 0", status: "pending" }],
});
const settings = makeSettings({ maxParallelSteps: 1 });
const failingSession = makeMockSession(async () => {
throw new Error("step execution failed");
});
failingSession.dispose = vi.fn(() => {
throw new Error("dispose failed");
});
mockedCreateKbAgent.mockResolvedValue({ session: failingSession } as any);
const executor = new StepSessionExecutor({
taskDetail: task,
worktreePath: "/project/.worktrees/main",
rootDir: "/project",
settings,
});
const resultsPromise = executor.executeAll();
await vi.advanceTimersByTimeAsync(60_000);
const results = await resultsPromise;
expect(results).toHaveLength(1);
expect(results[0]?.success).toBe(false);
expect(getStepSessionLogger().warn).toHaveBeenCalledWith(
expect.stringContaining("Failed to dispose session for step 0: dispose failed"),
);
});
});
describe("terminateAllSessions", () => {
it("disposes all active sessions and clears map", async () => {
const session0 = makeMockSession();

15
packages/engine/src/step-session-executor.ts
View File

@@ -983,8 +983,9 @@ export class StepSessionExecutor {
stuckTaskDetector?.untrackTask(trackingKey);
try {
session?.dispose();
} catch {
/* best-effort */
} catch (err: unknown) {
const msg = err instanceof Error ? err.message : String(err);
stepExecLog.warn(`Failed to dispose session for step ${stepIndex}: ${msg}`);
}
}
}
@@ -1163,8 +1164,9 @@ export class StepSessionExecutor {
{ cwd: worktreePath, encoding: "utf-8" },
);
commits = stdout.trim();
} catch {
stepExecLog.warn(`Could not list commits in parallel worktree for step ${stepIndex}`);
} catch (err: unknown) {
const msg = err instanceof Error ? err.message : String(err);
stepExecLog.warn(`Could not list commits in parallel worktree for step ${stepIndex}: ${msg}`);
return;
}
@@ -1188,8 +1190,9 @@ export class StepSessionExecutor {
// Cherry-pick conflict — abort and log
try {
await execAsync("git cherry-pick --abort", { cwd: primaryPath });
} catch {
// Ignore abort failure
} catch (abortErr: unknown) {
const msg = abortErr instanceof Error ? abortErr.message : String(abortErr);
stepExecLog.warn(`Cherry-pick --abort failed for step ${stepIndex}: ${msg}`);
}
throw new Error(
`Cherry-pick conflict for commit ${sha} in step ${stepIndex}: ${

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

@@ -11,6 +11,7 @@ import { join } from "node:path";
import { mkdir, writeFile, rm, mkdtemp } from "node:fs/promises";
import { tmpdir } from "node:os";
import { setTimeout as delay } from "node:timers/promises";
import { triageLog } from "./logger.js";
const { mockReviewStep, mockCreateKbAgent } = vi.hoisted(() => ({
mockReviewStep: vi.fn(),
@@ -2235,6 +2236,208 @@ describe("stuck task detector integration", () => {
});
});
describe("specifyTask — status restore failure diagnostics", () => {
it("logs warning when status restore fails during pause abort", async () => {
const warnSpy = vi.spyOn(triageLog, "warn");
const settingsListeners: Array<(e: any) => void> = [];
const store = {
on: vi.fn((event: string, cb: (e: any) => void) => {
if (event === "settings:updated") settingsListeners.push(cb);
}),
getTask: vi.fn().mockResolvedValue({ ...mockTaskDetail }),
getSettings: vi.fn().mockResolvedValue({ maxConcurrent: 2, maxWorktrees: 4, pollIntervalMs: 10000, groupOverlappingFiles: false, autoMerge: true } as Settings),
listTasks: vi.fn().mockResolvedValue([]),
updateTask: vi.fn().mockImplementation(async (_id: string, patch: Record<string, unknown>) => {
if (patch?.status === null) {
throw new Error("pause restore failed");
}
}),
logEntry: vi.fn().mockResolvedValue(undefined),
appendAgentLog: vi.fn().mockResolvedValue(undefined),
parseDependenciesFromPrompt: vi.fn().mockResolvedValue([]),
} as unknown as TaskStore;
let resolveDispose!: () => void;
const disposePromise = new Promise<void>((r) => { resolveDispose = r; });
mockCreateKbAgent.mockResolvedValue({
session: {
state: {},
sessionManager: {},
prompt: vi.fn().mockReturnValue(disposePromise),
dispose: vi.fn().mockImplementation(() => resolveDispose()),
navigateTree: vi.fn(),
},
});
const { promptWithFallback } = await import("./pi.js");
const promptWithFallbackMock = promptWithFallback as ReturnType<typeof vi.fn>;
promptWithFallbackMock.mockReturnValueOnce(disposePromise);
const task: Task = { id: "FN-001", description: "test", column: "triage", dependencies: [], steps: [], currentStep: 0, log: [], createdAt: "", updatedAt: "" };
const processor = new TriageProcessor(store, "/tmp/root");
const specifyPromise = processor.specifyTask(task);
await new Promise((r) => setTimeout(r, 20));
for (const listener of settingsListeners) {
listener({ settings: { globalPause: true }, previous: { globalPause: false } });
}
await expect(specifyPromise).resolves.toBeUndefined();
expect(warnSpy).toHaveBeenCalledWith(
expect.stringContaining("FN-001: failed to restore status to 'null' during pause-abort cleanup"),
);
warnSpy.mockRestore();
});
it("logs warning when status restore fails during stuck-detector abort", async () => {
const warnSpy = vi.spyOn(triageLog, "warn");
const store = {
on: vi.fn(),
getTask: vi.fn().mockResolvedValue({ ...mockTaskDetail }),
getSettings: vi.fn().mockResolvedValue({ maxConcurrent: 2, maxWorktrees: 4, pollIntervalMs: 10000, groupOverlappingFiles: false, autoMerge: true } as Settings),
listTasks: vi.fn().mockResolvedValue([]),
updateTask: vi.fn().mockImplementation(async (_id: string, patch: Record<string, unknown>) => {
if (patch?.status === null) {
throw new Error("stuck restore failed");
}
}),
logEntry: vi.fn().mockResolvedValue(undefined),
appendAgentLog: vi.fn().mockResolvedValue(undefined),
parseDependenciesFromPrompt: vi.fn().mockResolvedValue([]),
} as unknown as TaskStore;
let resolveDispose!: () => void;
const disposePromise = new Promise<void>((r) => { resolveDispose = r; });
const mockDispose = vi.fn().mockImplementation(() => resolveDispose());
mockCreateKbAgent.mockResolvedValue({
session: {
state: {},
sessionManager: {},
prompt: vi.fn().mockReturnValue(disposePromise),
dispose: mockDispose,
navigateTree: vi.fn(),
},
});
const { promptWithFallback } = await import("./pi.js");
const promptWithFallbackMock = promptWithFallback as ReturnType<typeof vi.fn>;
promptWithFallbackMock.mockReturnValueOnce(disposePromise);
const task: Task = { id: "FN-002", description: "test", column: "triage", dependencies: [], steps: [], currentStep: 0, log: [], createdAt: "", updatedAt: "" };
const processor = new TriageProcessor(store, "/tmp/root");
const specifyPromise = processor.specifyTask(task);
await new Promise((r) => setTimeout(r, 20));
processor.markStuckAborted("FN-002");
mockDispose();
await expect(specifyPromise).resolves.toBeUndefined();
expect(warnSpy).toHaveBeenCalledWith(
expect.stringContaining("FN-002: failed to restore status to 'null' during stuck-detector abort cleanup"),
);
warnSpy.mockRestore();
});
it("logs warning when logEntry fails during rate-limit retry", async () => {
vi.useFakeTimers();
const warnSpy = vi.spyOn(triageLog, "warn");
try {
const task: Task = {
id: "FN-207",
description: "Rate limit test",
column: "triage",
dependencies: [],
steps: [],
currentStep: 0,
log: [],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
} as Task;
const store = createMockStore({
getTask: vi.fn().mockResolvedValue({ ...task, attachments: [], comments: [] }),
logEntry: vi.fn().mockImplementation(async (_taskId: string, message: string) => {
if (message.includes("Rate limited — retry")) {
throw new Error("log write failed");
}
}),
});
mockCreateKbAgent.mockResolvedValue({
session: {
state: {},
sessionManager: { getLeafId: vi.fn().mockReturnValue(null) },
prompt: vi.fn().mockResolvedValue(undefined),
dispose: vi.fn(),
navigateTree: vi.fn(),
},
});
const { promptWithFallback } = await import("./pi.js");
const promptWithFallbackMock = promptWithFallback as ReturnType<typeof vi.fn>;
promptWithFallbackMock
.mockRejectedValueOnce(new Error("429 Too Many Requests"))
.mockResolvedValueOnce(undefined);
const processor = new TriageProcessor(store, "/test/root", {
pollIntervalMs: 100_000,
});
const specifyPromise = processor.specifyTask(task);
await vi.advanceTimersByTimeAsync(60_000);
await expect(specifyPromise).resolves.toBeUndefined();
expect(warnSpy).toHaveBeenCalledWith(
expect.stringContaining("FN-207: failed to log rate-limit retry entry"),
);
} finally {
warnSpy.mockRestore();
vi.useRealTimers();
}
});
it("logs warning when transient-error retry status update fails", async () => {
const warnSpy = vi.spyOn(triageLog, "warn");
const task: Task = {
id: "FN-208",
description: "Transient retry test",
column: "triage",
dependencies: [],
steps: [],
currentStep: 0,
log: [],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
} as Task;
const store = createMockStore({
getTask: vi.fn().mockResolvedValue({ ...task, attachments: [] }),
updateTask: vi.fn().mockImplementation(async (_id: string, patch: Record<string, unknown>) => {
if (patch?.recoveryRetryCount === 1) {
throw new Error("retry status update failed");
}
}),
});
mockCreateKbAgent.mockRejectedValueOnce(new Error("upstream connect error"));
const processor = new TriageProcessor(store, "/test/root", {
pollIntervalMs: 100_000,
});
await expect(processor.specifyTask(task)).resolves.toBeUndefined();
expect(warnSpy).toHaveBeenCalledWith(
expect.stringContaining("FN-208: failed to restore status to 'null' during transient-error retry scheduling"),
);
warnSpy.mockRestore();
});
});
describe("tool callback behavior (FN-1500)", () => {
it("records activity via stuckTaskDetector on tool callbacks", async () => {
const recordActivity = vi.fn();

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

@@ -464,7 +464,11 @@ export class TriageProcessor {
const settings = await this.store.getSettings();
const promptPath = join(this.rootDir, ".fusion", "tasks", task.id, "PROMPT.md");
const written = await readFile(promptPath, "utf-8").catch(() => "");
const written = await readFile(promptPath, "utf-8").catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
triageLog.warn(`${task.id}: failed to read PROMPT.md during approved-spec recovery (${promptPath}): ${msg}`);
return "";
});
if (!written.trim()) {
triageLog.warn(`${task.id} approved-spec recovery skipped — PROMPT.md missing or empty`);
@@ -682,8 +686,9 @@ export class TriageProcessor {
break;
}
}
} catch {
// Graceful fallback
} catch (err: unknown) {
const msg = err instanceof Error ? err.message : String(err);
triageLog.warn(`${task.id}: failed to resolve triage agent instructions, continuing with defaults: ${msg}`);
}
}
const triageSystemPrompt = buildSystemPromptWithInstructions(
@@ -806,7 +811,10 @@ export class TriageProcessor {
this.pauseAborted.delete(task.id);
triageLog.log(`${task.id} aborted by pause — clearing status`);
const restoreStatus = task.status === "needs-respecify" ? "needs-respecify" : null;
await this.store.updateTask(task.id, { status: restoreStatus }).catch(() => {});
await this.store.updateTask(task.id, { status: restoreStatus }).catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
triageLog.warn(`${task.id}: failed to restore status to '${restoreStatus}' during pause-abort cleanup: ${msg}`);
});
return;
}
@@ -814,7 +822,10 @@ export class TriageProcessor {
this.stuckAborted.delete(task.id);
triageLog.log(`${task.id} killed by stuck detector — clearing status for retry`);
const restoreStatus = task.status === "needs-respecify" ? "needs-respecify" : null;
await this.store.updateTask(task.id, { status: restoreStatus }).catch(() => {});
await this.store.updateTask(task.id, { status: restoreStatus }).catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
triageLog.warn(`${task.id}: failed to restore status to '${restoreStatus}' during stuck-detector abort cleanup: ${msg}`);
});
return;
}
@@ -979,7 +990,11 @@ export class TriageProcessor {
const written = await readFile(
join(this.rootDir, promptPath),
"utf-8",
).catch(() => "");
).catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
triageLog.warn(`${task.id}: failed to read generated PROMPT.md before finalization (${promptPath}): ${msg}`);
return "";
});
await this.finalizeApprovedTask(task, written, settings, {
isRespecify,
@@ -998,7 +1013,10 @@ export class TriageProcessor {
onRetry: (attempt, delayMs, error) => {
const delaySec = Math.round(delayMs / 1000);
triageLog.warn(`⏳ ${task.id} rate limited — retry ${attempt} in ${delaySec}s: ${error.message}`);
this.store.logEntry(task.id, `Rate limited — retry ${attempt} in ${delaySec}s`).catch(() => {});
this.store.logEntry(task.id, `Rate limited — retry ${attempt} in ${delaySec}s`).catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
triageLog.warn(`${task.id}: failed to log rate-limit retry entry: ${msg}`);
});
},
});
@@ -1020,14 +1038,20 @@ export class TriageProcessor {
// For re-specification, restore needs-respecify status; otherwise clear to null
// so the next poll can re-pick this task up.
const restoreStatus = task.status === "needs-respecify" ? "needs-respecify" : null;
await this.store.updateTask(task.id, { status: restoreStatus }).catch(() => {});
await this.store.updateTask(task.id, { status: restoreStatus }).catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
triageLog.warn(`${task.id}: failed to restore status to '${restoreStatus}' during pause-abort error cleanup: ${msg}`);
});
} else if (this.stuckAborted.has(task.id)) {
// Stuck task detector killed this session — clear specifying status so the
// next poll retries the task from scratch without reporting an error.
this.stuckAborted.delete(task.id);
triageLog.log(`${task.id} killed by stuck detector — clearing status for retry`);
const restoreStatus = task.status === "needs-respecify" ? "needs-respecify" : null;
await this.store.updateTask(task.id, { status: restoreStatus }).catch(() => {});
await this.store.updateTask(task.id, { status: restoreStatus }).catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
triageLog.warn(`${task.id}: failed to restore status to '${restoreStatus}' during stuck-detector error cleanup: ${msg}`);
});
} else {
// Check if the error is a usage-limit error and trigger global pause
if (this.options.usageLimitPauser && isUsageLimitError(errorMessage)) {
@@ -1049,32 +1073,47 @@ export class TriageProcessor {
// Silent transient errors (e.g., "request was aborted") are noisy — skip logging
if (!isSilentTransientError(errorMessage)) {
triageLog.warn(`⚡ ${task.id} transient error during triage — retry ${attempt}/${MAX_RECOVERY_RETRIES} in ${delay}: ${errorMessage}`);
await this.store.logEntry(task.id, `Transient error during specification (retry ${attempt}/${MAX_RECOVERY_RETRIES} in ${delay}): ${errorMessage}`).catch(() => {});
await this.store.logEntry(task.id, `Transient error during specification (retry ${attempt}/${MAX_RECOVERY_RETRIES} in ${delay}): ${errorMessage}`).catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
triageLog.warn(`${task.id}: failed to log transient-error retry entry: ${msg}`);
});
}
const restoreStatus = task.status === "needs-respecify" ? "needs-respecify" : null;
await this.store.updateTask(task.id, {
status: restoreStatus,
recoveryRetryCount: decision.nextState.recoveryRetryCount,
nextRecoveryAt: decision.nextState.nextRecoveryAt,
}).catch(() => {});
}).catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
triageLog.warn(`${task.id}: failed to restore status to '${restoreStatus}' during transient-error retry scheduling: ${msg}`);
});
return;
}
// Recovery budget exhausted — freeze in triage with error for manual intervention
triageLog.error(`✗ ${task.id} transient error retries exhausted (${MAX_RECOVERY_RETRIES} attempts): ${errorMessage}`);
await this.store.logEntry(task.id, `Specification failed after ${MAX_RECOVERY_RETRIES} transient errors: ${errorMessage}`).catch(() => {});
await this.store.logEntry(task.id, `Specification failed after ${MAX_RECOVERY_RETRIES} transient errors: ${errorMessage}`).catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
triageLog.warn(`${task.id}: failed to log transient-error retries-exhausted entry: ${msg}`);
});
await this.store.updateTask(task.id, {
error: `Specification failed after ${MAX_RECOVERY_RETRIES} transient errors: ${errorMessage}`,
recoveryRetryCount: null,
nextRecoveryAt: null,
}).catch(() => {});
}).catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
triageLog.warn(`${task.id}: failed to persist transient-error retries-exhausted state: ${msg}`);
});
this.options.onSpecifyError?.(task, err instanceof Error ? err : new Error(errorMessage));
return;
}
// For re-specification, restore needs-respecify status so it can be retried;
// otherwise clear to null so the next poll can re-pick the task up.
const restoreStatus = task.status === "needs-respecify" ? "needs-respecify" : null;
await this.store.updateTask(task.id, { status: restoreStatus }).catch(() => {});
await this.store.updateTask(task.id, { status: restoreStatus }).catch((restoreErr: unknown) => {
const msg = restoreErr instanceof Error ? restoreErr.message : String(restoreErr);
triageLog.warn(`${task.id}: failed to restore status to '${restoreStatus}' after specification error: ${msg}`);
});
triageLog.error(`✗ ${task.id} specification failed:`, errorMessage);
this.options.onSpecifyError?.(task, err instanceof Error ? err : new Error(errorMessage));
}
@@ -1160,7 +1199,9 @@ export class TriageProcessor {
content: [{ type: "text" as const, text: parts.join("\n") }],
details: {},
};
} catch {
} catch (err: unknown) {
const msg = err instanceof Error ? err.message : String(err);
triageLog.warn(`${options.parentTaskId}: task_get lookup failed for ${params.id}: ${msg}`);
return {
content: [
{ type: "text" as const, text: `Task ${params.id} not found.` },
@@ -1192,7 +1233,9 @@ export class TriageProcessor {
let parentTask: Awaited<ReturnType<typeof store.getTask>> | undefined;
try {
parentTask = await store.getTask(options.parentTaskId);
} catch {
} catch (err: unknown) {
const msg = err instanceof Error ? err.message : String(err);
triageLog.warn(`${options.parentTaskId}: failed to load parent task for task_create inheritance: ${msg}`);
// Parent task not found or error - proceed without inheritance
parentTask = undefined;
}
@@ -1296,7 +1339,11 @@ export class TriageProcessor {
const promptContent = await readFile(
join(rootDir, promptPath),
"utf-8",
).catch(() => "");
).catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
triageLog.warn(`${taskId}: failed to read PROMPT.md for review_spec (${promptPath}): ${msg}`);
return "";
});
if (!promptContent) {
return {
@@ -1384,7 +1431,9 @@ export class TriageProcessor {
triageLog.log(
`${taskId}: RETHINK — session rewound to checkpoint ${checkpointId}`,
);
} catch {
} catch (rewindErr: unknown) {
const msg = rewindErr instanceof Error ? rewindErr.message : String(rewindErr);
triageLog.warn(`${taskId}: RETHINK navigateTree rewind failed, falling back to branchWithSummary: ${msg}`);
// Fallback to branchWithSummary
try {
sessionRef.current.sessionManager.branchWithSummary(
@@ -1596,7 +1645,9 @@ export async function readAttachmentContents(
text,
});
}
} catch {
} catch (err: unknown) {
const msg = err instanceof Error ? err.message : String(err);
triageLog.warn(`${taskId}: failed to read attachment '${att.filename}', skipping: ${msg}`);
// Skip unreadable attachments
continue;
}