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

fix(engine,droid-cli): stop subagents with parent task and harden droid-cli probes

Browse Source 
engine: reviewer subagents previously kept running after the parent task
was moved out of in-progress, paused, or globally paused — they spawn
their own AgentSession outside `activeSessions`/`activeStepExecutors`,
so the existing kill paths never reached them. Track them in a per-task
`activeSubagentSessions` map (mirrored in TriageProcessor) and dispose
on the same triggers as the main session. ReviewOptions gains
`onSessionCreated` / `onSessionEnded` callbacks the executor and triage
processor wire to register/unregister.

droid-cli: probe timeouts (`validateCliPresence`, `validateCliAuth`,
`runDroidProbe`) raised from 5s to 45s — observed cold-start is ~20s,
so 5s reported the binary as missing even when present. Provider gains
a `FIRST_LINE_TIMEOUT_MS` (60s) cold-start guard so a hung droid binary
is reported with an actionable error instead of being indistinguishable
from a slow-thinking turn. Fix the await race in `streamViaCli`: when
SIGKILL destroys stdout mid-buffer, `rl` may never emit "close", so the
promise also resolves on `proc.close` and forces `rl.close()` — prevents
the engine's "executor did not unwind within 60s — hung subprocess".

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsxdsm
2026-05-01 06:44:07 -07:00
parent d6d591b532
commit 6032b900a4
6 changed files with 238 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
packages/engine/src/__tests__/reviewer.test.ts
View File

@@ -920,3 +920,57 @@ describe("reviewStep — skill selection resolver contract (FN-1510/FN-1511)", (
expect(opts.skillSelection?.sessionPurpose).toBe("reviewer");
});
});
describe("reviewStep — subagent lifecycle hooks", () => {
beforeEach(() => {
vi.clearAllMocks();
});
it("fires onSessionCreated then onSessionEnded with the same session, in order", async () => {
const mockSession = createMockSession("### Verdict: APPROVE\n### Summary\nOk.");
mockedCreateFnAgent.mockResolvedValue(mockSession);
const events: Array<{ type: "created" | "ended"; sameSession: boolean }> = [];
const onSessionCreated = vi.fn((s: any) => {
events.push({ type: "created", sameSession: s === mockSession.session });
});
const onSessionEnded = vi.fn((s: any) => {
events.push({ type: "ended", sameSession: s === mockSession.session });
});
await reviewStep(
"/tmp/worktree", "FN-200", 1, "Hook test", "plan", "# prompt",
undefined,
{ onSessionCreated, onSessionEnded },
);
expect(onSessionCreated).toHaveBeenCalledTimes(1);
expect(onSessionEnded).toHaveBeenCalledTimes(1);
expect(events).toEqual([
{ type: "created", sameSession: true },
{ type: "ended", sameSession: true },
]);
expect(mockSession.session.dispose).toHaveBeenCalledTimes(1);
});
it("fires onSessionEnded even when promptWithFallback throws", async () => {
const mockSession = createMockSession("");
mockedCreateFnAgent.mockResolvedValue(mockSession);
const { promptWithFallback } = await import("../pi.js");
vi.mocked(promptWithFallback).mockRejectedValueOnce(new Error("boom"));
const onSessionCreated = vi.fn();
const onSessionEnded = vi.fn();
await expect(
reviewStep(
"/tmp/worktree", "FN-201", 1, "Error path", "plan", "# prompt",
undefined,
{ onSessionCreated, onSessionEnded },
),
).rejects.toThrow("boom");
expect(onSessionCreated).toHaveBeenCalledTimes(1);
expect(onSessionEnded).toHaveBeenCalledTimes(1);
});
});

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

@@ -563,6 +563,15 @@ export class TaskExecutor {
}>();
/** Active step-session executors per task (mutually exclusive with activeSessions). */
private activeStepExecutors = new Map<string, StepSessionExecutor>();
/**
* Reviewer subagent sessions per task. Reviewers (`reviewer.ts`) create their
* own AgentSessions that aren't part of `activeSessions`/`activeStepExecutors`,
* so without this map they survive when the parent task is stopped — they
* keep producing log entries and step transitions after the user thinks they
* killed the task. Disposed alongside the main session in the move-out,
* pause, and global-pause handlers below.
*/
private activeSubagentSessions = new Map<string, Set<AgentSession>>();
/** Tasks that were paused mid-execution (to avoid marking them as "failed"). */
private pausedAborted = new Set<string>();
/** Tasks that had a dependency added mid-execution (abort + discard worktree). */
@@ -638,6 +647,52 @@ export class TaskExecutor {
* Sessions are not disposed here so any near-complete agent loop still has a
* chance to wrap up during the runtime's graceful drain window.
*/
/**
* Register a subagent session (e.g. reviewer) under its parent task ID so it
* can be disposed when the parent stops. Used as the `onSessionCreated`
* callback passed to `reviewStep`.
*/
private registerSubagentSession(taskId: string, session: AgentSession): void {
let set = this.activeSubagentSessions.get(taskId);
if (!set) {
set = new Set();
this.activeSubagentSessions.set(taskId, set);
}
set.add(session);
}
/**
* Deregister a subagent session that has finished naturally. The reviewer's
* own `finally` block disposes the session — this just removes it from the
* map.
*/
private unregisterSubagentSession(taskId: string, session: AgentSession): void {
const set = this.activeSubagentSessions.get(taskId);
if (!set) return;
set.delete(session);
if (set.size === 0) this.activeSubagentSessions.delete(taskId);
}
/**
* Dispose all subagent sessions for a task and remove them from the map.
* Called by the kill paths (move-out-of-in-progress, pause, global pause)
* so subagents stop alongside the main session.
*/
private disposeSubagentsForTask(taskId: string, reason: string): void {
const set = this.activeSubagentSessions.get(taskId);
if (!set || set.size === 0) return;
executorLog.log(`${taskId}: disposing ${set.size} subagent session(s) — ${reason}`);
for (const session of set) {
try {
session.dispose();
} catch (err) {
executorLog.warn(`${taskId}: failed to dispose subagent session: ${err}`);
}
}
this.activeSubagentSessions.delete(taskId);
}
abortAllSessionBash(): void {
for (const [taskId, { session }] of this.activeSessions) {
try {
@@ -713,6 +768,12 @@ export class TaskExecutor {
);
this.activeStepExecutors.delete(task.id);
}
// Reviewer subagents run in their own sessions outside `activeSessions`
// and `activeStepExecutors`, so the loops above don't reach them.
// Without this, a reviewer keeps running (and emitting verdicts that
// trigger step transitions) even after the parent task was moved out
// of in-progress.
this.disposeSubagentsForTask(task.id, `parent moved from in-progress to ${to}`);
// Clean up all in-memory state for this task so nothing leaks across runs.
// This prevents zombie state from persisting when a task moves away from
// in-progress while execute() is still unwinding, or when the scheduler
@@ -746,6 +807,7 @@ export class TaskExecutor {
this.loopRecoveryState.delete(task.id);
this.spawnedAgents.delete(task.id);
this.stuckAborted.delete(task.id);
this.disposeSubagentsForTask(task.id, "task paused");
return;
}
if (task.paused && this.activeStepExecutors.has(task.id)) {
@@ -758,6 +820,7 @@ export class TaskExecutor {
this.loopRecoveryState.delete(task.id);
this.spawnedAgents.delete(task.id);
this.stuckAborted.delete(task.id);
this.disposeSubagentsForTask(task.id, "task paused");
return;
}
@@ -895,6 +958,12 @@ export class TaskExecutor {
// When globalPause transitions from false → true, terminate all active agent sessions.
store.on("settings:updated", ({ settings, previous }) => {
if (settings.globalPause && !previous.globalPause) {
// Dispose every reviewer subagent across every task. The per-task loops
// below handle main + step sessions; reviewers live in their own map
// and would otherwise outlive the global pause.
for (const taskId of [...this.activeSubagentSessions.keys()]) {
this.disposeSubagentsForTask(taskId, "global pause");
}
for (const [taskId, { session }] of this.activeSessions) {
executorLog.log(`Global pause — terminating agent session for ${taskId}`);
this.pausedAborted.add(taskId);
@@ -3535,6 +3604,11 @@ export class TaskExecutor {
agentStore: this.options.agentStore,
rootDir: this.rootDir,
settings,
// Track the reviewer's session under this task so it's disposed
// alongside the main session when the task moves out of
// in-progress, is paused, or the engine globally pauses.
onSessionCreated: (s) => this.registerSubagentSession(taskId, s),
onSessionEnded: (s) => this.unregisterSubagentSession(taskId, s),
},
);
const result = sem

20
packages/engine/src/reviewer.ts
View File

@@ -268,6 +268,19 @@ export interface ReviewOptions {
settings?: Settings;
/** Plugin runner for runtime selection. When provided, enables plugin runtime lookup. */
pluginRunner?: import("./plugin-runner.js").PluginRunner;
/**
* Fired immediately after the reviewer's `AgentSession` is created. The
* caller can register the session in a per-task subagent map so that the
* session can be disposed when the parent task moves out of `in-progress`,
* is paused, or the engine globally pauses. Without this hook, reviewer
* sessions outlive their parent task on a stop signal.
*/
onSessionCreated?: (session: import("@mariozechner/pi-coding-agent").AgentSession) => void;
/**
* Fired in a `finally` block after the reviewer is fully done (or aborted).
* Pair with `onSessionCreated` to deregister from the subagent map.
*/
onSessionEnded?: (session: import("@mariozechner/pi-coding-agent").AgentSession) => void;
}
/**
@@ -514,6 +527,12 @@ export async function reviewStep(
await options.store.logEntry(options.taskId, `Reviewer using model: ${describeModel(session)}`);
}
// Notify the caller so it can track this session in a per-task subagent map.
// If the parent task is later moved out of in-progress, paused, or the engine
// is globally paused, the caller will dispose this session — preventing the
// reviewer from outliving its parent task.
options.onSessionCreated?.(session);
let reviewText = "";
// Capture the reviewer's full text output (still needed for verdict extraction)
@@ -533,6 +552,7 @@ export async function reviewStep(
} finally {
if (agentLogger) await agentLogger.flush();
session.dispose();
options.onSessionEnded?.(session);
}
const verdict = extractVerdict(reviewText);

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

@@ -489,6 +489,13 @@ export class TriageProcessor {
private wasEnginePaused = false;
/** Active agent sessions per task, used to terminate on pause. */
private activeSessions = new Map<string, { dispose: () => void }>();
/**
* Reviewer subagent sessions per task. The spec reviewer (`reviewer.ts`)
* creates its own AgentSession that isn't part of `activeSessions`, so
* without this map it survives a global pause and continues producing
* verdicts. Mirrors `TaskExecutor.activeSubagentSessions`.
*/
private activeSubagentSessions = new Map<string, Set<AgentSession>>();
/** Tasks aborted due to globalPause (to avoid reporting as errors). */
private pauseAborted = new Set<string>();
/** Tasks killed by the stuck task detector (to avoid reporting as errors). */
@@ -512,6 +519,11 @@ 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}`,
@@ -612,6 +624,43 @@ export class TriageProcessor {
this.stuckAborted.add(taskId);
}
/**
* Register a reviewer subagent session under its parent task. Used as the
* `onSessionCreated` callback passed to `reviewStep`. Mirrors the
* TaskExecutor implementation.
*/
private registerSubagentSession(taskId: string, session: AgentSession): void {
let set = this.activeSubagentSessions.get(taskId);
if (!set) {
set = new Set();
this.activeSubagentSessions.set(taskId, set);
}
set.add(session);
}
/** Deregister a reviewer subagent that finished naturally. */
private unregisterSubagentSession(taskId: string, session: AgentSession): void {
const set = this.activeSubagentSessions.get(taskId);
if (!set) return;
set.delete(session);
if (set.size === 0) this.activeSubagentSessions.delete(taskId);
}
/** Dispose all reviewer subagents for a task and remove them from the map. */
private disposeSubagentsForTask(taskId: string, reason: string): void {
const set = this.activeSubagentSessions.get(taskId);
if (!set || set.size === 0) return;
planLog.log(`${taskId}: disposing ${set.size} subagent session(s) — ${reason}`);
for (const session of set) {
try {
session.dispose();
} catch (err) {
planLog.warn(`${taskId}: failed to dispose subagent session: ${err}`);
}
}
this.activeSubagentSessions.delete(taskId);
}
/**
* Return a snapshot of tasks currently being specified by this processor.
* Used by self-healing maintenance to avoid recovering live sessions.
@@ -1772,6 +1821,10 @@ export class TriageProcessor {
userComments: currentUserComments.length > 0 ? currentUserComments : undefined,
agentStore: this.options.agentStore,
rootDir,
// Track the spec reviewer's session under this task so it's
// disposed alongside the main triage session on global pause.
onSessionCreated: (s) => this.registerSubagentSession(taskId, s),
onSessionEnded: (s) => this.unregisterSubagentSession(taskId, s),
},
);
const result = sem