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

fix(FN-1285): harden async listener error guards

Browse Source 
- Wrap TaskExecutor task:updated async listener logic in a top-level guard and log uncaught listener failures
- Add explicit catch handling for ProjectManager activity logging and dashboard stuck-detector settings-triggered checkNow calls
- Document async EventEmitter guard conventions in scheduler and hybrid executor listener wiring
- Add regression tests for executor and dashboard listener guards and include a patch changeset for @gsxdsm/fusion
This commit is contained in:
gsxdsm
2026-04-08 11:49:33 -07:00
parent d3569361d9
commit 9ac5b19646
8 changed files with 227 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -110,6 +110,7 @@ import type { Column, Task, TaskDetail } from "@fusion/core";
import { SessionManager } from "@mariozechner/pi-coding-agent";
import { StuckTaskDetector } from "./stuck-task-detector.js";
import { StepSessionExecutor } from "./step-session-executor.js";
import { executorLog } from "./logger.js";
const mockedCreateHaiAgent = vi.mocked(createKbAgent);
const mockedSessionManager = vi.mocked(SessionManager);
@@ -3011,6 +3012,41 @@ describe("TaskExecutor executor model hot-swap", () => {
});
});
describe("TaskExecutor task:updated listener guards", () => {
it("catches and logs errors from async task:updated operations", async () => {
const store = createMockStore();
const terminateError = new Error("terminate failed");
const terminateAllSessions = vi.fn().mockRejectedValue(terminateError);
const executor = new TaskExecutor(store, "/tmp/test");
(executor as any).activeStepExecutors.set("FN-001", {
terminateAllSessions,
});
const taskUpdatedHandler = (store.on as unknown as ReturnType<typeof vi.fn>).mock.calls
.find((call: any[]) => call[0] === "task:updated")?.[1];
expect(taskUpdatedHandler).toBeTypeOf("function");
await expect(taskUpdatedHandler({
id: "FN-001",
title: "Guard test",
description: "Guard test",
column: "in-progress",
paused: true,
dependencies: [],
steps: [],
currentStep: 0,
log: [],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
} satisfies Task)).resolves.toBeUndefined();
expect(terminateAllSessions).toHaveBeenCalledTimes(1);
expect(executorLog.error).toHaveBeenCalledWith("Uncaught error in task:updated listener:", terminateError);
});
});
describe("TaskExecutor global pause behavior", () => {
beforeEach(() => {
vi.clearAllMocks();

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

@@ -329,117 +329,121 @@ export class TaskExecutor {
// 4. Comments are marked as seen BEFORE injection to prevent retry loops on failure
// 5. Each injection is logged to the task for user visibility
store.on("task:updated", async (task) => {
// Handle pause - terminate the agent session or step sessions
if (task.paused && this.activeSessions.has(task.id)) {
executorLog.log(`Pausing ${task.id} — terminating agent session`);
this.pausedAborted.add(task.id);
this.options.stuckTaskDetector?.untrackTask(task.id);
const { session } = this.activeSessions.get(task.id)!;
session.dispose();
return;
}
if (task.paused && this.activeStepExecutors.has(task.id)) {
executorLog.log(`Pausing ${task.id} — terminating step sessions`);
this.pausedAborted.add(task.id);
this.options.stuckTaskDetector?.untrackTask(task.id);
const stepExecutor = this.activeStepExecutors.get(task.id)!;
await stepExecutor.terminateAllSessions();
return;
}
// Handle unpause of an in-progress task with no active session.
// This covers orphaned states (e.g., engine restarted while task was
// paused in-progress) where the task needs to resume execution.
// The executing/executing guards prevent duplicate runs.
if (!task.paused && task.column === "in-progress" && !this.activeSessions.has(task.id)) {
if (!this.executing.has(task.id)) {
executorLog.log(`Unpaused ${task.id} in-progress with no session — resuming execution`);
try {
await this.clearResumeFailureState(task);
await this.store.logEntry(task.id, "Resuming execution after unpause");
} catch { /* non-critical */ }
this.execute(task).catch((err) =>
executorLog.error(`Failed to resume unpaused ${task.id}:`, err),
);
try {
// Handle pause - terminate the agent session or step sessions
if (task.paused && this.activeSessions.has(task.id)) {
executorLog.log(`Pausing ${task.id} — terminating agent session`);
this.pausedAborted.add(task.id);
this.options.stuckTaskDetector?.untrackTask(task.id);
const { session } = this.activeSessions.get(task.id)!;
session.dispose();
return;
}
if (task.paused && this.activeStepExecutors.has(task.id)) {
executorLog.log(`Pausing ${task.id} — terminating step sessions`);
this.pausedAborted.add(task.id);
this.options.stuckTaskDetector?.untrackTask(task.id);
const stepExecutor = this.activeStepExecutors.get(task.id)!;
await stepExecutor.terminateAllSessions();
return;
}
return;
}
// Handle executor model hot-swap on active single-session executions
if (this.activeSessions.has(task.id) && !task.paused) {
const activeEntry = this.activeSessions.get(task.id)!;
const providerChanged = task.modelProvider !== activeEntry.lastModelProvider;
const modelIdChanged = task.modelId !== activeEntry.lastModelId;
if (providerChanged || modelIdChanged) {
activeEntry.lastModelProvider = task.modelProvider;
activeEntry.lastModelId = task.modelId;
const settings = await this.store.getSettings();
const newProvider = task.modelProvider && task.modelId
? task.modelProvider
: settings?.defaultProvider;
const newModelId = task.modelProvider && task.modelId
? task.modelId
: settings?.defaultModelId;
if (newProvider && newModelId) {
// Handle unpause of an in-progress task with no active session.
// This covers orphaned states (e.g., engine restarted while task was
// paused in-progress) where the task needs to resume execution.
// The executing/executing guards prevent duplicate runs.
if (!task.paused && task.column === "in-progress" && !this.activeSessions.has(task.id)) {
if (!this.executing.has(task.id)) {
executorLog.log(`Unpaused ${task.id} in-progress with no session — resuming execution`);
try {
const model = this.modelRegistry.find(newProvider, newModelId);
if (model) {
await activeEntry.session.setModel(model);
executorLog.log(`${task.id}: executor model hot-swapped to ${newProvider}/${newModelId}`);
await this.store.logEntry(task.id, `Model changed to ${newProvider}/${newModelId}`);
} else {
executorLog.log(`${task.id}: model ${newProvider}/${newModelId} not found in registry for hot-swap`);
await this.clearResumeFailureState(task);
await this.store.logEntry(task.id, "Resuming execution after unpause");
} catch { /* non-critical */ }
this.execute(task).catch((err) =>
executorLog.error(`Failed to resume unpaused ${task.id}:`, err),
);
}
return;
}
// Handle executor model hot-swap on active single-session executions
if (this.activeSessions.has(task.id) && !task.paused) {
const activeEntry = this.activeSessions.get(task.id)!;
const providerChanged = task.modelProvider !== activeEntry.lastModelProvider;
const modelIdChanged = task.modelId !== activeEntry.lastModelId;
if (providerChanged || modelIdChanged) {
activeEntry.lastModelProvider = task.modelProvider;
activeEntry.lastModelId = task.modelId;
const settings = await this.store.getSettings();
const newProvider = task.modelProvider && task.modelId
? task.modelProvider
: settings?.defaultProvider;
const newModelId = task.modelProvider && task.modelId
? task.modelId
: settings?.defaultModelId;
if (newProvider && newModelId) {
try {
const model = this.modelRegistry.find(newProvider, newModelId);
if (model) {
await activeEntry.session.setModel(model);
executorLog.log(`${task.id}: executor model hot-swapped to ${newProvider}/${newModelId}`);
await this.store.logEntry(task.id, `Model changed to ${newProvider}/${newModelId}`);
} else {
executorLog.log(`${task.id}: model ${newProvider}/${newModelId} not found in registry for hot-swap`);
}
} catch (err: any) {
executorLog.error(`${task.id}: failed to hot-swap model: ${err.message}`);
await this.store.logEntry(task.id, `Model change failed: ${err.message}`);
}
} catch (err: any) {
executorLog.error(`${task.id}: failed to hot-swap model: ${err.message}`);
await this.store.logEntry(task.id, `Model change failed: ${err.message}`);
}
}
}
}
// Handle steering comments - inject new ones into the running session
// Only process if session is active (activeSessions check is sufficient
// since entries are only added when a task is in-progress)
if (this.activeSessions.has(task.id) && task.steeringComments) {
const activeSession = this.activeSessions.get(task.id)!;
const { session, seenSteeringIds } = activeSession;
// Handle steering comments - inject new ones into the running session
// Only process if session is active (activeSessions check is sufficient
// since entries are only added when a task is in-progress)
if (this.activeSessions.has(task.id) && task.steeringComments) {
const activeSession = this.activeSessions.get(task.id)!;
const { session, seenSteeringIds } = activeSession;
// Find new steering comments that haven't been seen yet
const newComments = task.steeringComments.filter(c => !seenSteeringIds.has(c.id));
// Find new steering comments that haven't been seen yet
const newComments = task.steeringComments.filter(c => !seenSteeringIds.has(c.id));
if (newComments.length > 0) {
for (const comment of newComments) {
const summary = comment.text.length > 80
? comment.text.slice(0, 80) + "..."
: comment.text;
if (newComments.length > 0) {
for (const comment of newComments) {
const summary = comment.text.length > 80
? comment.text.slice(0, 80) + "..."
: comment.text;
// Mark as seen BEFORE attempting injection to prevent retry loops on failure
seenSteeringIds.add(comment.id);
// Mark as seen BEFORE attempting injection to prevent retry loops on failure
seenSteeringIds.add(comment.id);
// Format and inject the comment
const commentMessage = formatCommentForInjection(comment);
try {
executorLog.log(`Injecting comment into ${task.id}: ${summary}`);
await session.steer(commentMessage);
executorLog.log(`Successfully injected comment into ${task.id}`);
// Format and inject the comment
const commentMessage = formatCommentForInjection(comment);
try {
executorLog.log(`Injecting comment into ${task.id}: ${summary}`);
await session.steer(commentMessage);
executorLog.log(`Successfully injected comment into ${task.id}`);
// Log to the task that comment was received
await this.store.logEntry(
task.id,
`Comment received mid-execution: ${summary}`,
`by ${comment.author}`
);
} catch (err) {
executorLog.error(`Failed to inject comment for ${task.id}:`, err);
// Comment is already marked as seen - we won't retry to avoid spamming
// the agent with failed injections. The error is logged for debugging.
// Log to the task that comment was received
await this.store.logEntry(
task.id,
`Comment received mid-execution: ${summary}`,
`by ${comment.author}`
);
} catch (err) {
executorLog.error(`Failed to inject comment for ${task.id}:`, err);
// Comment is already marked as seen - we won't retry to avoid spamming
// the agent with failed injections. The error is logged for debugging.
}
}
}
}
} catch (err) {
executorLog.error("Uncaught error in task:updated listener:", err);
}
});

5
packages/engine/src/hybrid-executor.ts
View File

@@ -429,6 +429,11 @@ export class HybridExecutor extends EventEmitter<HybridExecutorEvents> {
/**
* Set up listeners for CentralCore project events.
*
* Async guard convention: any async operation triggered from these listeners
* must end with an explicit `.catch(...)` handler (for example, project
* removal calls `removeProject(...).catch(...)`) so EventEmitter dispatch
* cannot surface unhandled promise rejections.
*/
private setupCentralCoreListeners(): void {
// When a new project is registered, we don't auto-add it

12
packages/engine/src/project-manager.ts
View File

@@ -352,7 +352,9 @@ export class ProjectManager extends EventEmitter<ProjectManagerEvents> {
// Forward task:created
runtime.on("task:created", (task: Task) => {
this.emit("task:created", { projectId, projectName, task });
this.logActivity("task:created", projectId, projectName, `Task ${task.id} created`, task.id, task.title);
this.logActivity("task:created", projectId, projectName, `Task ${task.id} created`, task.id, task.title).catch((err: unknown) => {
projectManagerLog.warn(`Failed to log task:created activity for ${projectId}:`, err);
});
});
// Forward task:moved
@@ -366,7 +368,9 @@ export class ProjectManager extends EventEmitter<ProjectManagerEvents> {
data.task.id,
data.task.title,
{ from: data.from, to: data.to }
);
).catch((err: unknown) => {
projectManagerLog.warn(`Failed to log task:moved activity for ${projectId}:`, err);
});
});
// Forward task:updated
@@ -382,7 +386,9 @@ export class ProjectManager extends EventEmitter<ProjectManagerEvents> {
projectId,
projectName,
`Error in ${projectName}: ${error.message}`
);
).catch((err: unknown) => {
projectManagerLog.warn(`Failed to log task:failed activity for ${projectId}:`, err);
});
});
// Forward health changes

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

@@ -110,6 +110,13 @@ export class Scheduler {
/** Tracks mission-linked tasks observed with status=failed before moveTask clears status/error. */
private failedTaskIds = new Set<string>();
/**
* Async listener guard convention:
* - Any async mission helper invoked from event listeners is wrapped in internal try/catch
* (`handleMissionTaskStart` / `handleMissionTaskCompletion`).
* - Fire-and-forget Promise chains in listeners terminate with `.catch(...)`.
* Keep this invariant when adding new async EventEmitter callbacks.
*/
constructor(
private store: TaskStore,
private options: SchedulerOptions = {},