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

fix(engine): stop tears down in-progress merger and triager sessions

Browse Source 
TriageProcessor.stop() previously only halted the polling loop, so
in-flight specify sessions and their reviewer subagents kept streaming
past shutdown. Extracted the existing global-pause teardown into
abortAndDisposeActiveSessions() and call it from stop() too.

aiMergeTask creates three sessions during a merge — autostash resolver,
in-merge verification fix agent, and pull-rebase conflict resolver — but
only the autostash one was registered via onSession. The other two are
now registered (with onSession threaded through pushToRemoteAfterMerge
into the rebase resolver chain), so ProjectEngine.stop() actually
disposes whichever merger session is running when shutdown lands.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsxdsm
2026-05-06 00:04:13 -07:00
parent f65f1a16a3
commit b2aed0fd42
3 changed files with 81 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
.changeset/engine-stop-aborts-triage-sessions.md Normal file
View File

@@ -0,0 +1,19 @@
---
"@runfusion/fusion": patch
---
Engine stop now tears down in-progress merger and triager agent sessions
that previously kept streaming past shutdown.
**Triager**: `TriageProcessor.stop()` previously only halted the polling
loop, leaving any in-flight specify session and its reviewer subagents
streaming LLM tokens and tool calls past shutdown. It now aborts and
disposes them via the same path the global-pause handler uses.
**Merger**: `aiMergeTask` creates up to three distinct agent sessions
during a merge — autostash conflict resolver, in-merge verification fix
agent, and pull-rebase conflict resolver — but only the autostash session
was registered via `onSession` for the engine to track. The fix-agent and
rebase-resolver sessions are now also registered, so
`ProjectEngine.stop()` actually disposes whichever merger session is
running when shutdown lands.

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

@@ -785,6 +785,10 @@ Do not refactor, rename broadly, or make opportunistic improvements.
taskTitle: taskForSkillContext?.title,
}),
});
// Register so engine.stop() can dispose this session — without this the
// fix agent keeps streaming past shutdown because it's not the autostash
// session that the engine tracks.
options.onSession?.(session);
const runId = mergeRunContext?.runId;
const agentId = mergeRunContext?.agentId ?? "merger";
@@ -2725,6 +2729,7 @@ async function resolveComplexRebaseConflictsWithAi(
pluginRunner?: import("./plugin-runner.js").PluginRunner;
signal?: AbortSignal;
runtimeHint?: string;
onSession?: (session: { dispose: () => void }) => void;
},
): Promise<void> {
mergerLog.log(`${taskId}: resolving ${conflictedFiles.length} complex rebase conflict(s) with AI`);
@@ -2780,6 +2785,10 @@ You are assisting with a paused \`git pull --rebase\`.
taskId,
}),
});
// Register so engine.stop() can dispose this session — without this, an
// in-progress rebase conflict resolution keeps streaming past shutdown
// (the engine only tracks the autostash session by default).
options?.onSession?.(session);
const prompt = [
`Resolve rebase conflicts for task ${taskId}.`,
@@ -2814,7 +2823,12 @@ async function resolveRebaseConflictSet(
rootDir: string,
taskId: string,
settings: Settings,
options?: { onAgentText?: (delta: string) => void; signal?: AbortSignal; runtimeHint?: string },
options?: {
onAgentText?: (delta: string) => void;
signal?: AbortSignal;
runtimeHint?: string;
onSession?: (session: { dispose: () => void }) => void;
},
): Promise<void> {
const conflictedFiles = await getConflictedFiles(rootDir);
if (conflictedFiles.length === 0) return;
@@ -2857,7 +2871,12 @@ async function pullWithRebaseAndResolveConflicts(
settings: Settings,
remote: string,
branch: string,
options?: { onAgentText?: (delta: string) => void; signal?: AbortSignal; runtimeHint?: string },
options?: {
onAgentText?: (delta: string) => void;
signal?: AbortSignal;
runtimeHint?: string;
onSession?: (session: { dispose: () => void }) => void;
},
): Promise<void> {
const pullCommand = `git pull --rebase ${quoteArg(remote)} ${quoteArg(branch)}`;
try {
@@ -2948,7 +2967,12 @@ export async function pushToRemoteAfterMerge(
rootDir: string,
taskId: string,
settings: Settings,
options?: { onAgentText?: (delta: string) => void; signal?: AbortSignal; runtimeHint?: string },
options?: {
onAgentText?: (delta: string) => void;
signal?: AbortSignal;
runtimeHint?: string;
onSession?: (session: { dispose: () => void }) => void;
},
): Promise<{ pushed: boolean; error?: string }> {
let target: { remote: string; branch: string };
@@ -4412,6 +4436,7 @@ export async function aiMergeTask(
onAgentText: options.onAgentText,
signal: options.signal,
runtimeHint: pushRuntimeHint,
onSession: options.onSession,
});
if (pushResult.pushed) {
mergerLog.log(`${taskId}: pushed merged result to remote`);

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

@@ -525,27 +525,7 @@ export class TriageProcessor {
// When globalPause transitions from false → true, terminate all active triage sessions.
store.on("settings:updated", ({ settings, previous }) => {
if (settings.globalPause && !previous.globalPause) {
// Dispose every reviewer subagent first so they don't keep streaming
// verdicts while the main triage session is being torn down.
for (const taskId of [...this.activeSubagentSessions.keys()]) {
this.disposeSubagentsForTask(taskId, "global pause");
}
for (const [taskId, session] of this.activeSessions) {
planLog.log(
`Global pause — terminating triage session for ${taskId}`,
);
this.pauseAborted.add(taskId);
this.options.stuckTaskDetector?.untrackTask(taskId);
// abort() interrupts any in-flight LLM stream / tool call;
// dispose() then releases session resources.
const sessionWithAbort = session as { abort?: () => Promise<void>; dispose: () => void };
if (typeof sessionWithAbort.abort === "function") {
void sessionWithAbort.abort().catch((err) => {
planLog.warn(`Failed to abort triage session for ${taskId}: ${err}`);
});
}
session.dispose();
}
this.abortAndDisposeActiveSessions("global pause");
}
});
@@ -618,9 +598,42 @@ export class TriageProcessor {
this.pollInterval = null;
this.activePollMs = null;
}
// Tear down any in-flight specify sessions and reviewer subagents so they
// don't keep streaming LLM tokens / tool calls past engine shutdown.
this.abortAndDisposeActiveSessions("engine stop");
planLog.log("Processor stopped");
}
/**
* Abort and dispose every active specify session and reviewer subagent.
* Used by the global-pause handler and by `stop()`.
*
* Reviewer subagents are torn down first so they don't keep streaming
* verdicts while the main triage session is being disposed. abort()
* interrupts any in-flight LLM stream / tool call; dispose() then
* releases session resources.
*/
private abortAndDisposeActiveSessions(reason: string): void {
for (const taskId of [...this.activeSubagentSessions.keys()]) {
this.disposeSubagentsForTask(taskId, reason);
}
for (const [taskId, session] of this.activeSessions) {
planLog.log(`${reason} — terminating triage session for ${taskId}`);
this.pauseAborted.add(taskId);
this.options.stuckTaskDetector?.untrackTask(taskId);
const sessionWithAbort = session as {
abort?: () => Promise<void>;
dispose: () => void;
};
if (typeof sessionWithAbort.abort === "function") {
void sessionWithAbort.abort().catch((err) => {
planLog.warn(`Failed to abort triage session for ${taskId}: ${err}`);
});
}
session.dispose();
}
}
/**
* Mark a task as stuck-aborted so the catch block knows not to treat
* the disposed session as a genuine failure.