diff --git a/.changeset/pause-triage-planning.md b/.changeset/pause-triage-planning.md new file mode 100644 index 0000000000..9eb9815977 --- /dev/null +++ b/.changeset/pause-triage-planning.md @@ -0,0 +1,5 @@ +--- +"@runfusion/fusion": patch +--- + +Respect per-task pause state during triage planning so paused tasks do not auto-advance after specification approval. diff --git a/packages/engine/src/__tests__/triage-pause-abort.test.ts b/packages/engine/src/__tests__/triage-pause-abort.test.ts new file mode 100644 index 0000000000..48486e2be3 --- /dev/null +++ b/packages/engine/src/__tests__/triage-pause-abort.test.ts @@ -0,0 +1,237 @@ +import "./executor-test-helpers.js"; +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { Settings, Task, TaskStore } from "@fusion/core"; + +import { TriageProcessor } from "../triage.js"; +import { resetExecutorMocks } from "./executor-test-helpers.js"; + +type Listener = (...args: any[]) => void; + +function createEventedStore(overrides: Record = {}) { + const listeners = new Map>(); + const store = { + getSettings: vi.fn().mockResolvedValue({ pollIntervalMs: 60_000, maxConcurrent: 1, maxWorktrees: 1, autoMerge: true }), + listTasks: vi.fn().mockResolvedValue([]), + updateTask: vi.fn().mockResolvedValue(undefined), + moveTask: vi.fn().mockResolvedValue(undefined), + on: vi.fn((event: string, listener: Listener) => { + const set = listeners.get(event) ?? new Set(); + set.add(listener); + listeners.set(event, set); + }), + off: vi.fn((event: string, listener: Listener) => { + listeners.get(event)?.delete(listener); + }), + ...overrides, + } as any; + + return { + store, + emit(event: string, ...args: any[]) { + for (const listener of listeners.get(event) ?? []) { + listener(...args); + } + }, + }; +} + +function createFinalizeStore(overrides: Partial = {}): TaskStore { + return { + listTasks: vi.fn().mockResolvedValue([]), + getTask: vi.fn().mockResolvedValue(createTask()), + getSettings: vi.fn().mockResolvedValue({ requirePlanApproval: false } as Settings), + parseDependenciesFromPrompt: vi.fn().mockResolvedValue([]), + parseStepsFromPrompt: vi.fn().mockResolvedValue([]), + parseFileScopeFromPrompt: vi.fn().mockResolvedValue([]), + updateTask: vi.fn().mockResolvedValue(undefined), + moveTask: vi.fn().mockResolvedValue(undefined), + logEntry: vi.fn().mockResolvedValue(undefined), + deleteTask: vi.fn().mockResolvedValue(undefined), + on: vi.fn(), + off: vi.fn(), + ...overrides, + } as unknown as TaskStore; +} + +function createTask(overrides: Partial = {}): Task { + return { + id: "FN-PAUSE-1", + title: "Paused planning task", + description: "desc", + column: "triage", + status: "planning", + dependencies: [], + steps: [], + currentStep: 0, + log: [{ timestamp: new Date().toISOString(), action: "Spec review: APPROVE" }], + createdAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + ...overrides, + } as Task; +} + +describe("TriageProcessor per-task pause aborts", () => { + beforeEach(() => { + resetExecutorMocks(); + vi.clearAllMocks(); + }); + + it("does not start planning work for an already-paused triage task", async () => { + const task = createTask({ id: "FN-PAUSE-START", paused: true, status: null }); + const { store } = createEventedStore({ listTasks: vi.fn().mockResolvedValue([task]) }); + const processor = new TriageProcessor(store, "/tmp/root"); + const specifyTask = vi.spyOn(processor as any, "specifyTask").mockResolvedValue(undefined); + + (processor as any).running = true; + await (processor as any).poll(); + + expect(specifyTask).not.toHaveBeenCalled(); + expect((processor as any).processing.has(task.id)).toBe(false); + }); + + it("aborts and disposes an active specify session on task:updated pause without moving to todo", async () => { + const { store, emit } = createEventedStore(); + const stuckTaskDetector = { untrackTask: vi.fn() }; + const processor = new TriageProcessor(store, "/tmp/root", { stuckTaskDetector } as any); + const abort = vi.fn().mockResolvedValue(undefined); + const dispose = vi.fn(); + + processor.start(); + (processor as any).activeSessions.set("FN-PAUSE-2", { abort, dispose }); + + emit("task:updated", { id: "FN-PAUSE-2", paused: true }); + await Promise.resolve(); + + expect(abort).toHaveBeenCalledTimes(1); + expect(dispose).toHaveBeenCalledTimes(1); + expect((processor as any).activeSessions.has("FN-PAUSE-2")).toBe(false); + expect((processor as any).pauseAborted.has("FN-PAUSE-2")).toBe(true); + expect(stuckTaskDetector.untrackTask).toHaveBeenCalledWith("FN-PAUSE-2"); + expect(store.moveTask).not.toHaveBeenCalled(); + + processor.stop(); + }); + + it("treats userPaused task updates as pause aborts", async () => { + const { store, emit } = createEventedStore(); + const processor = new TriageProcessor(store, "/tmp/root"); + const abort = vi.fn().mockResolvedValue(undefined); + const dispose = vi.fn(); + + processor.start(); + (processor as any).activeSessions.set("FN-USER-PAUSE", { abort, dispose }); + + emit("task:updated", { id: "FN-USER-PAUSE", userPaused: true }); + await Promise.resolve(); + + expect(abort).toHaveBeenCalledTimes(1); + expect(dispose).toHaveBeenCalledTimes(1); + expect((processor as any).pauseAborted.has("FN-USER-PAUSE")).toBe(true); + + processor.stop(); + }); + + it("does not abort on non-paused updates or paused ids with no active session", () => { + const { store, emit } = createEventedStore(); + const processor = new TriageProcessor(store, "/tmp/root"); + const abort = vi.fn().mockResolvedValue(undefined); + const dispose = vi.fn(); + + processor.start(); + (processor as any).activeSessions.set("FN-ACTIVE", { abort, dispose }); + + expect(() => emit("task:updated", { id: "FN-ACTIVE", paused: false })).not.toThrow(); + expect(() => emit("task:updated", { id: "FN-MISSING", paused: true })).not.toThrow(); + + expect(abort).not.toHaveBeenCalled(); + expect(dispose).not.toHaveBeenCalled(); + expect((processor as any).activeSessions.has("FN-ACTIVE")).toBe(true); + + processor.stop(); + }); + + it("detaches the task:updated pause listener on stop", () => { + const { store, emit } = createEventedStore(); + const processor = new TriageProcessor(store, "/tmp/root"); + const abort = vi.fn().mockResolvedValue(undefined); + const dispose = vi.fn(); + + processor.start(); + (processor as any).activeSessions.set("FN-PAUSE-STOP", { abort, dispose }); + processor.stop(); + const abortCallsAfterStop = abort.mock.calls.length; + const disposeCallsAfterStop = dispose.mock.calls.length; + + emit("task:updated", { id: "FN-PAUSE-STOP", paused: true }); + + expect(abort).toHaveBeenCalledTimes(abortCallsAfterStop); + expect(dispose).toHaveBeenCalledTimes(disposeCallsAfterStop); + }); +}); + +describe("TriageProcessor paused finalization guard", () => { + beforeEach(() => { + resetExecutorMocks(); + vi.clearAllMocks(); + }); + + it("does not move an approved task to todo when the re-read task is paused", async () => { + const task = createTask({ id: "FN-FINALIZE-PAUSED" }); + const store = createFinalizeStore({ getTask: vi.fn().mockResolvedValue({ ...task, paused: true }) }); + const processor = new TriageProcessor(store, "/tmp/root"); + + await (processor as any).finalizeApprovedTask( + task, + "# Task: FN-FINALIZE-PAUSED\n\n## File Scope\n- packages/engine/src/triage.ts\n", + { requirePlanApproval: false } as Settings, + ); + + expect(store.moveTask).not.toHaveBeenCalled(); + expect(store.updateTask).toHaveBeenLastCalledWith(task.id, { status: null }); + expect(store.logEntry).toHaveBeenCalledWith( + task.id, + "Specification approved but task is paused — leaving in triage, will resume on unpause", + ); + }); + + it("does not move to awaiting-approval when the re-read task is userPaused", async () => { + const task = createTask({ id: "FN-FINALIZE-USER-PAUSED" }); + const store = createFinalizeStore({ getTask: vi.fn().mockResolvedValue({ ...task, userPaused: true }) }); + const processor = new TriageProcessor(store, "/tmp/root"); + + await (processor as any).finalizeApprovedTask( + task, + "# Task: FN-FINALIZE-USER-PAUSED\n\n## File Scope\n- packages/engine/src/triage.ts\n", + { requirePlanApproval: true } as Settings, + ); + + expect(store.moveTask).not.toHaveBeenCalled(); + expect(store.updateTask).not.toHaveBeenCalledWith(task.id, expect.objectContaining({ status: "awaiting-approval" })); + expect(store.updateTask).toHaveBeenLastCalledWith(task.id, { status: null }); + }); + + it("keeps the unpaused approved-spec happy path moving to todo", async () => { + const task = createTask({ id: "FN-FINALIZE-HAPPY" }); + const store = createFinalizeStore({ getTask: vi.fn().mockResolvedValue({ ...task, paused: false, userPaused: false }) }); + const processor = new TriageProcessor(store, "/tmp/root"); + + await (processor as any).finalizeApprovedTask( + task, + "# Task: FN-FINALIZE-HAPPY\n\n## File Scope\n- packages/engine/src/triage.ts\n", + { requirePlanApproval: false } as Settings, + ); + + expect(store.moveTask).toHaveBeenCalledWith(task.id, "todo"); + }); + + it("does not recover an approved planning task while it is paused", async () => { + const task = createTask({ id: "FN-RECOVER-PAUSED", paused: true }); + const store = createFinalizeStore(); + const processor = new TriageProcessor(store, "/tmp/root"); + + await expect(processor.recoverApprovedTask(task)).resolves.toBe(false); + + expect(store.moveTask).not.toHaveBeenCalled(); + expect(store.updateTask).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/engine/src/triage.ts b/packages/engine/src/triage.ts index 34948329fb..212cbaf04f 100644 --- a/packages/engine/src/triage.ts +++ b/packages/engine/src/triage.ts @@ -146,6 +146,7 @@ export class TriageProcessor { /** Tasks killed by the stuck task detector (to avoid reporting as errors). */ private stuckAborted = new Set(); private taskDeletedHandler?: (task: Task) => void; + private taskPausedHandler?: (task: Task) => void; /** * @param store — Task store instance (also used to listen for `settings:updated` events) @@ -218,6 +219,32 @@ export class TriageProcessor { this.activeSessions.delete(task.id); } }; + + this.taskPausedHandler = (task: Task) => { + if (!task?.id || (task.paused !== true && task.userPaused !== true)) { + return; + } + if (this.activeSubagentSessions.has(task.id)) { + this.disposeSubagentsForTask(task.id, "task paused"); + } + if (this.activeSessions.has(task.id)) { + const session = this.activeSessions.get(task.id)!; + planLog.log(`task paused — terminating triage session for ${task.id}`); + this.pauseAborted.add(task.id); + this.options.stuckTaskDetector?.untrackTask(task.id); + const sessionWithAbort = session as { + abort?: () => Promise; + dispose: () => void; + }; + if (typeof sessionWithAbort.abort === "function") { + void sessionWithAbort.abort().catch((err) => { + planLog.warn(`Failed to abort triage session for ${task.id}: ${err}`); + }); + } + session.dispose(); + this.activeSessions.delete(task.id); + } + }; } start(): void { @@ -226,6 +253,9 @@ export class TriageProcessor { if (this.taskDeletedHandler && typeof this.store.on === "function") { this.store.on("task:deleted", this.taskDeletedHandler); } + if (this.taskPausedHandler && typeof this.store.on === "function") { + this.store.on("task:updated", this.taskPausedHandler); + } // Clear stale "planning" statuses left by a prior crash/restart. // No triage agent is actually running at startup, so any task still @@ -267,6 +297,9 @@ export class TriageProcessor { if (this.taskDeletedHandler && typeof this.store.off === "function") { this.store.off("task:deleted", this.taskDeletedHandler); } + if (this.taskPausedHandler && typeof this.store.off === "function") { + this.store.off("task:updated", this.taskPausedHandler); + } // 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"); @@ -407,6 +440,11 @@ export class TriageProcessor { return false; } + if (task.paused === true || task.userPaused === true) { + planLog.log(`${task.id} approved-spec recovery skipped — task is paused`); + return false; + } + if (!hasLatestSpecReviewApproval(task)) { return false; } @@ -2244,6 +2282,25 @@ export class TriageProcessor { planLog.warn(`${task.id}: near-duplicate backstop failed open: ${message}`); } + let latestTransitionTask: Task | undefined; + try { + latestTransitionTask = await this.store.getTask(task.id); + } catch (err: unknown) { + const message = err instanceof Error ? err.message : String(err); + planLog.warn(`${task.id}: failed to re-read task before approved-spec transition (${message}); proceeding with original task snapshot`); + latestTransitionTask = task; + } + if (latestTransitionTask?.paused === true || latestTransitionTask?.userPaused === true) { + const restoreStatus = options.isReplan ? "needs-replan" : null; + await this.store.updateTask(task.id, { status: restoreStatus }); + await this.store.logEntry( + task.id, + "Specification approved but task is paused — leaving in triage, will resume on unpause", + ); + planLog.log(`${task.id} approved specification paused — leaving in triage, will resume on unpause`); + return; + } + if (settings.requirePlanApproval) { const approvalUpdates: Record = { status: "awaiting-approval" }; if (shouldApplyPromptDeclaredTitle && promptDeclaredTitle) {