diff --git a/packages/engine/src/__tests__/executor-archive-releases-active-session.test.ts b/packages/engine/src/__tests__/executor-archive-releases-active-session.test.ts index b67931032c..0f815b5b16 100644 --- a/packages/engine/src/__tests__/executor-archive-releases-active-session.test.ts +++ b/packages/engine/src/__tests__/executor-archive-releases-active-session.test.ts @@ -165,3 +165,54 @@ describe("archiving a task releases its active-session registry entries (FN-7717 ).not.toThrow(); }); }); + +describe("workflow graph column boundaries do not abort their own run", () => { + it.each(["triage", "in-review"] as const)( + "keeps the graph runner alive across in-progress → %s", + async (to) => { + const { executor, store } = makeExecutor(); + const task = { id: `TASK-GRAPH-${to}`, column: to } as any; + const abortSpy = vi + .spyOn(executor as any, "awaitAbortInFlightTaskWork") + .mockResolvedValue(undefined); + (store as any).moveTask = vi.fn(async (_taskId: string, column: string) => { + store.emit("task:moved", { + task: { ...task, column }, + from: "in-progress", + to: column, + source: "engine", + }); + }); + + (executor as any).graphRouting.add(task.id); + try { + await (executor as any).buildColumnBoundaryHooks(task).moveTask(to, { + fromColumn: "in-progress", + nodeId: to === "triage" ? "plan-replan" : "review", + }); + await Promise.resolve(); + + expect(abortSpy).not.toHaveBeenCalled(); + } finally { + (executor as any).graphRouting.delete(task.id); + } + }, + ); + + it("still aborts an external engine move away from in-progress", async () => { + const { executor, store } = makeExecutor(); + const abortSpy = vi + .spyOn(executor as any, "awaitAbortInFlightTaskWork") + .mockResolvedValue(undefined); + + store.emit("task:moved", { + task: { id: "TASK-EXTERNAL", column: "todo" }, + from: "in-progress", + to: "todo", + source: "engine", + }); + await Promise.resolve(); + + expect(abortSpy).toHaveBeenCalledOnce(); + }); +}); diff --git a/packages/engine/src/__tests__/self-healing-advanced-triage.test.ts b/packages/engine/src/__tests__/self-healing-advanced-triage.test.ts new file mode 100644 index 0000000000..8faf1234b2 --- /dev/null +++ b/packages/engine/src/__tests__/self-healing-advanced-triage.test.ts @@ -0,0 +1,167 @@ +import { EventEmitter } from "node:events"; +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { Settings, Task, TaskStore } from "@fusion/core"; +import { activeSessionRegistry } from "../active-session-registry.js"; +import { SelfHealingManager } from "../self-healing.js"; + +function task(id: string, overrides: Partial = {}): Task { + return { + id, + title: id, + description: id, + column: "triage", + status: null, + paused: false, + worktree: `/tmp/${id}`, + workflowIrPinNodeId: "code-review-remediation", + workflowIrPinColumnId: "in-progress", + dependencies: [], + steps: [{ name: "Fix review", status: "pending" }], + currentStep: 0, + log: [], + createdAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + ...overrides, + } as Task; +} + +function storeFor(tasks: Task[]): TaskStore & EventEmitter { + const rows = new Map(tasks.map((entry) => [entry.id, entry])); + return Object.assign(new EventEmitter(), { + getSettings: vi.fn(async () => ({ globalPause: false, enginePaused: false }) as Settings), + listTasks: vi.fn(async () => [...rows.values()]), + getTask: vi.fn(async (id: string) => rows.get(id)), + moveTask: vi.fn(async (id: string, column: Task["column"]) => { + const current = rows.get(id)!; + rows.set(id, { ...current, column }); + return rows.get(id); + }), + moveTaskIf: vi.fn(async ( + id: string, + column: Task["column"], + predicate: (live: Task) => boolean | Promise, + ) => { + const current = rows.get(id)!; + if (!await predicate(current)) return { task: current, moved: false }; + const moved = { ...current, column } as Task; + rows.set(id, moved); + return { task: moved, moved: true }; + }), + logEntry: vi.fn(async () => undefined), + }) as unknown as TaskStore & EventEmitter; +} + +describe("advanced workflow tasks stranded in triage", () => { + beforeEach(() => activeSessionRegistry.clear()); + + it("resumes incomplete remediation at its durable pinned column", async () => { + const stranded = task("FN-INCOMPLETE"); + const store = storeFor([stranded]); + const manager = new SelfHealingManager(store, { + rootDir: "/repo", + getExecutingTaskIds: () => new Set(), + }); + + expect(await manager.recoverAdvancedTriageTasks()).toBe(1); + expect(store.moveTaskIf).toHaveBeenCalledWith( + stranded.id, + "in-progress", + expect.any(Function), + expect.objectContaining({ + recoveryRehome: true, + preserveProgress: true, + preserveWorktree: true, + workflowMoveSource: "self-healing-advanced-triage", + }), + ); + }); + + it("promotes completed pinned work through the normal completion recovery seam", async () => { + const stranded = task("FN-COMPLETE", { + workflowIrPinNodeId: "merge", + workflowIrPinColumnId: undefined, + steps: [{ name: "Implement", status: "done" }], + }); + const store = storeFor([stranded]); + const recoverCompletedTask = vi.fn(async () => true); + const manager = new SelfHealingManager(store, { + rootDir: "/repo", + recoverCompletedTask, + getExecutingTaskIds: () => new Set(), + }); + + expect(await manager.recoverAdvancedTriageTasks()).toBe(1); + expect(recoverCompletedTask).toHaveBeenCalledWith(stranded); + expect(store.moveTaskIf).not.toHaveBeenCalled(); + }); + + it("leaves ordinary planning rows and actively-owned graph runs untouched", async () => { + const ordinary = task("FN-ORDINARY", { worktree: undefined, workflowIrPinNodeId: undefined }); + const active = task("FN-ACTIVE"); + const store = storeFor([ordinary, active]); + const manager = new SelfHealingManager(store, { + rootDir: "/repo", + getExecutingTaskIds: () => new Set([active.id]), + }); + + expect(await manager.recoverAdvancedTriageTasks()).toBe(0); + expect(store.moveTaskIf).not.toHaveBeenCalled(); + }); + + it("does not rehome a task claimed after discovery but before the atomic move", async () => { + const stranded = task("FN-RACING-CLAIM"); + const store = storeFor([stranded]); + let claimed = false; + const moveTaskIfMock = vi.mocked(store.moveTaskIf); + const moveTaskIf = moveTaskIfMock.getMockImplementation()!; + moveTaskIfMock.mockImplementation(async (...args) => { + claimed = true; + return moveTaskIf(...args); + }); + const manager = new SelfHealingManager(store, { + rootDir: "/repo", + getExecutingTaskIds: () => claimed ? new Set([stranded.id]) : new Set(), + }); + + expect(await manager.recoverAdvancedTriageTasks()).toBe(0); + expect(store.moveTaskIf).toHaveBeenCalledOnce(); + }); + + it("does not promote completed work when planning wins the ownership reservation", async () => { + const stranded = task("FN-COMPLETED-RACING-CLAIM", { + workflowIrPinNodeId: "merge", + workflowIrPinColumnId: undefined, + steps: [{ name: "Implement", status: "done" }], + }); + const store = storeFor([stranded]); + const recoverCompletedTask = vi.fn(async () => true); + const manager = new SelfHealingManager(store, { + rootDir: "/repo", + recoverCompletedTask, + getExecutingTaskIds: () => new Set(), + reserveAdvancedTriageRecovery: () => undefined, + }); + + expect(await manager.recoverAdvancedTriageTasks()).toBe(0); + expect(recoverCompletedTask).not.toHaveBeenCalled(); + }); + + it("releases the planning fence after completed recovery", async () => { + const stranded = task("FN-COMPLETED-RESERVED", { + workflowIrPinNodeId: "merge", + workflowIrPinColumnId: undefined, + steps: [{ name: "Implement", status: "done" }], + }); + const store = storeFor([stranded]); + const release = vi.fn(); + const manager = new SelfHealingManager(store, { + rootDir: "/repo", + recoverCompletedTask: vi.fn(async () => true), + getExecutingTaskIds: () => new Set(), + reserveAdvancedTriageRecovery: () => release, + }); + + expect(await manager.recoverAdvancedTriageTasks()).toBe(1); + expect(release).toHaveBeenCalledOnce(); + }); +}); diff --git a/packages/engine/src/__tests__/self-healing-db-corruption.test.ts b/packages/engine/src/__tests__/self-healing-db-corruption.test.ts index ab07bc9f4a..156bfe8f9e 100644 --- a/packages/engine/src/__tests__/self-healing-db-corruption.test.ts +++ b/packages/engine/src/__tests__/self-healing-db-corruption.test.ts @@ -49,6 +49,7 @@ const BATCH1_METHODS = [ const BATCH2_METHODS = [ "recoverCompletedTasks", "recoverStrandedCompletedTodoTasks", + "recoverAdvancedTriageTasks", "recoverStaleIncompleteReviewTasks", "recoverReviewTasksWithFailedPreMergeSteps", "recoverInterruptedMergingTasks", diff --git a/packages/engine/src/__tests__/triage.test.ts b/packages/engine/src/__tests__/triage.test.ts index 9f87c2143d..2d576e8b0d 100644 --- a/packages/engine/src/__tests__/triage.test.ts +++ b/packages/engine/src/__tests__/triage.test.ts @@ -1536,6 +1536,20 @@ describe("TriageProcessor", () => { expect(processor).toBeInstanceOf(TriageProcessor); }); + it("uses a synchronous reservation to keep planning and advanced recovery mutually exclusive", async () => { + const task = createTriageTask({ id: "FN-RECOVERY-RESERVED" }); + const release = processor.tryReserveAdvancedRecovery(task.id); + expect(release).toBeTypeOf("function"); + expect(processor.getProcessingTaskIds()).toContain(task.id); + expect(processor.getPlanningTaskIds()).not.toContain(task.id); + + await processor.specifyTask(task); + expect(store.getTask).not.toHaveBeenCalled(); + + release?.(); + expect(processor.getProcessingTaskIds()).not.toContain(task.id); + }); + /* FNXC:OriginalDescriptionInPrompt 2026-07-14-23:35: finalizeApprovedTask must inject ## Original Description with the task description @@ -1730,6 +1744,33 @@ Planner rewrote mission without the raw request. expect(specifySpy).toHaveBeenCalledWith(expect.objectContaining({ id: "FN-200" })); }); + it("does not repeatedly dispatch a triage row that already has executor advancement evidence", async () => { + const tasks: Task[] = [ + createTriageTask({ id: "FN-ADVANCED", worktree: "/tmp/fusion-fn-advanced" }), + createTriageTask({ id: "FN-UNPLANNED" }), + ]; + const triageStore = createMockStore({ + listTasks: vi.fn().mockResolvedValue(tasks), + getSettings: vi.fn().mockResolvedValue({ + maxConcurrent: 10, + maxTriageConcurrent: 10, + pollIntervalMs: 10_000, + groupOverlappingFiles: false, + autoMerge: true, + }), + }); + const triageProcessor = new TriageProcessor(triageStore, rootDir); + const specifySpy = vi + .spyOn(triageProcessor, "specifyTask") + .mockResolvedValue(undefined); + + (triageProcessor as any).running = true; + await (triageProcessor as any).poll(); + + expect(specifySpy).toHaveBeenCalledOnce(); + expect(specifySpy).toHaveBeenCalledWith(expect.objectContaining({ id: "FN-UNPLANNED" })); + }); + /* FNXC:GlobalConcurrencyControls 2026-07-14-18:30: When an in-progress executor already counts toward the live running-agent total, triage must leave room under the global cap instead of filling maxTriageConcurrent purely from semaphore.availableCount. diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 226d74ff56..f3721eaa4a 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -1660,6 +1660,13 @@ export class TaskExecutor { * Prevents the task:moved handler from dispatching execute() before the * bounce finishes its own dispatch. */ private workflowRerunPending = new Set(); + /** + * Task ids whose current `task:moved` event is being emitted by this + * executor's workflow column-boundary hook. The store emits synchronously, + * so this narrowly distinguishes a graph's own transition from an external + * engine/user move that must still hard-cancel the active run. + */ + private workflowBoundaryMovesInFlight = new Set(); /** FN-5256: in-flight session-disposal promises keyed by taskId. The * task:moved (away from in-progress) and task:deleted listeners populate * this so a fast re-dispatch (task:moved → in-progress) awaits the prior @@ -3038,6 +3045,12 @@ export class TaskExecutor { }), ); } else if (from === "in-progress") { + if (this.workflowBoundaryMovesInFlight.has(task.id) && this.graphRouting.has(task.id)) { + executorLog.log( + `[event:task:moved] Preserving graph run for ${task.id} across its own ${from} → ${to} boundary`, + ); + return; + } this.trackTaskDisposal( task.id, this.awaitAbortInFlightTaskWork(task.id, `parent moved from in-progress to ${to}`, { @@ -5823,13 +5836,18 @@ export class TaskExecutor { // row fields so an ordinary requeue re-resolves the CURRENT IR fresh. clearPin: pinPersistence.clearPin, moveTask: async (toColumn, ctx) => { - await this.store.moveTask(task.id, toColumn, { - moveSource: "engine", - workflowMoveSource: "workflow-graph", - bypassGuards: true, - preserveProgress: true, - workflowMoveMetadata: { fromColumn: ctx.fromColumn, nodeId: ctx.nodeId }, - }); + this.workflowBoundaryMovesInFlight.add(task.id); + try { + await this.store.moveTask(task.id, toColumn, { + moveSource: "engine", + workflowMoveSource: "workflow-graph", + bypassGuards: true, + preserveProgress: true, + workflowMoveMetadata: { fromColumn: ctx.fromColumn, nodeId: ctx.nodeId }, + }); + } finally { + this.workflowBoundaryMovesInFlight.delete(task.id); + } }, emitAudit: async (event) => { await this.store.recordRunAuditEvent?.({ diff --git a/packages/engine/src/runtimes/in-process-runtime.ts b/packages/engine/src/runtimes/in-process-runtime.ts index 6a1602785b..9207b76024 100644 --- a/packages/engine/src/runtimes/in-process-runtime.ts +++ b/packages/engine/src/runtimes/in-process-runtime.ts @@ -1049,7 +1049,8 @@ export class InProcessRuntime clearPhantomExecutorBinding: (taskId: string, options?: { preserveWorktrees?: boolean }) => this.executor?.clearPhantomExecutorBinding(taskId, options), listWorktreeHolders: () => this.executor?.listWorktreeHolders() ?? [], recoverApprovedTriageTask: (task) => this.triageProcessor?.recoverApprovedTask(task) ?? Promise.resolve(false), - getPlanningTaskIds: () => this.triageProcessor?.getProcessingTaskIds() ?? new Set(), + getPlanningTaskIds: () => this.triageProcessor?.getPlanningTaskIds() ?? new Set(), + reserveAdvancedTriageRecovery: (taskId) => this.triageProcessor?.tryReserveAdvancedRecovery(taskId), evictStaleTriageProcessing: () => this.triageProcessor?.evictStaleProcessing() ?? new Set(), enqueueMerge: this.mergeEnqueuer ? (taskId: string) => this.mergeEnqueuer?.(taskId) ?? false : undefined, requeueForAutoMerge: this.mergeEnqueuer ? (taskId: string) => this.mergeEnqueuer?.(taskId) ?? false : undefined, diff --git a/packages/engine/src/self-healing.ts b/packages/engine/src/self-healing.ts index b27cb8e6c1..731001c82f 100644 --- a/packages/engine/src/self-healing.ts +++ b/packages/engine/src/self-healing.ts @@ -296,6 +296,8 @@ export interface SelfHealingOptions { * Used to avoid recovering active triage sessions. */ getPlanningTaskIds?: () => Set; + /** Atomically fence planner ownership while advanced triage recovery runs. */ + reserveAdvancedTriageRecovery?: (taskId: string) => (() => void) | undefined; /** * Evict tasks from the triage processor's `processing` set that have been * there longer than the staleness threshold (hung promises from stuck kills). @@ -1377,6 +1379,7 @@ export class SelfHealingManager { { name: "no-progress-no-task-done", fn: () => this.recoverNoProgressNoTaskDoneFailures().then(() => undefined) }, { name: "completed-tasks", fn: () => this.recoverCompletedTasks().then(() => undefined) }, { name: "recover-stranded-completed-todo", fn: () => this.recoverStrandedCompletedTodoTasks().then(() => undefined) }, + { name: "recover-advanced-triage", fn: () => this.recoverAdvancedTriageTasks().then(() => undefined) }, { name: "stale-incomplete-review", fn: () => this.recoverStaleIncompleteReviewTasks().then(() => undefined) }, { name: "failed-pre-merge-steps", fn: () => this.recoverReviewTasksWithFailedPreMergeSteps().then(() => undefined) }, { name: "missing-worktree-review-failures", fn: () => this.recoverMissingWorktreeReviewFailures().then(() => undefined) }, @@ -2672,6 +2675,7 @@ export class SelfHealingManager { }, { name: "recover-completed-tasks", fn: () => this.recoverCompletedTasks() }, { name: "recover-stranded-completed-todo", fn: () => this.recoverStrandedCompletedTodoTasks() }, + { name: "recover-advanced-triage", fn: () => this.recoverAdvancedTriageTasks() }, { name: "recover-stale-incomplete-review", fn: () => this.recoverStaleIncompleteReviewTasks() }, { name: "recover-failed-pre-merge-steps", fn: () => this.recoverReviewTasksWithFailedPreMergeSteps() }, { name: "recover-missing-worktree-review-failures", fn: () => this.recoverMissingWorktreeReviewFailures() }, @@ -3009,6 +3013,111 @@ export class SelfHealingManager { } } + /** + * Re-home workflow-graph tasks stranded in the planner column after the + * executor aborted its own column-boundary move. A worktree plus a durable + * graph pin is the proof that this is advanced execution state, not an + * ordinary triage card. Completed work goes through the normal review + * recovery seam; incomplete remediation resumes at its pinned column. + */ + async recoverAdvancedTriageTasks(): Promise { + try { + const settings = await this.store.getSettings(); + if (settings.globalPause || settings.enginePaused) return 0; + + const tasks = await this.store.listTasks({ column: "triage", slim: true }); + const executingIds = this.options.getExecutingTaskIds?.() ?? new Set(); + const planningIds = this.options.getPlanningTaskIds?.() ?? new Set(); + const candidates = tasks.filter((task) => + task.column === "triage" + && task.status == null + && !task.paused + && !task.error + && Boolean(task.worktree) + && Boolean(task.workflowIrPinNodeId) + && !executingIds.has(task.id) + && !planningIds.has(task.id) + && this.options.isTaskActive?.(task.id) !== true + && !(task.worktree && activeSessionRegistry.isPathActive(task.worktree)), + ); + + let recovered = 0; + for (const snapshot of candidates) { + const releaseReservation = this.options.reserveAdvancedTriageRecovery?.(snapshot.id); + if (this.options.reserveAdvancedTriageRecovery && !releaseReservation) continue; + try { + const live = await this.store.getTask(snapshot.id); + if ( + live.column !== "triage" + || live.status != null + || live.paused + || live.error + || !live.worktree + || !live.workflowIrPinNodeId + || (this.options.getExecutingTaskIds?.() ?? new Set()).has(live.id) + || this.options.isTaskActive?.(live.id) === true + || activeSessionRegistry.isPathActive(live.worktree) + ) { + continue; + } + + const steps = live.steps ?? []; + const complete = steps.length > 0 + && steps.every((step) => step.status === "done" || step.status === "skipped"); + if (complete) { + if (!this.options.recoverCompletedTask) continue; + if (await this.options.recoverCompletedTask(live)) recovered++; + continue; + } + + const resumeColumn = live.workflowIrPinColumnId; + if (!resumeColumn || resumeColumn === "triage") continue; + const moved = await this.store.moveTaskIf(live.id, resumeColumn, (current) => + current.column === "triage" + && current.status == null + && !current.paused + && !current.error + && current.worktree === live.worktree + && current.workflowIrPinNodeId === live.workflowIrPinNodeId + && current.workflowIrPinColumnId === resumeColumn + && !(this.options.getExecutingTaskIds?.() ?? new Set()).has(current.id) + && !(this.options.getPlanningTaskIds?.() ?? new Set()).has(current.id) + && this.options.isTaskActive?.(current.id) !== true + && !(current.worktree && activeSessionRegistry.isPathActive(current.worktree)), + { + moveSource: "engine", + workflowMoveSource: "self-healing-advanced-triage", + workflowMoveMetadata: { + reason: "graph-boundary-self-abort-recovery", + pinnedNodeId: live.workflowIrPinNodeId, + }, + recoveryRehome: true, + bypassGuards: true, + preserveProgress: true, + preserveWorktree: true, + preserveResumeState: true, + }); + if (!moved.moved) continue; + await this.store.logEntry( + live.id, + `Auto-recovered workflow task stranded in triage — resumed at pinned ${resumeColumn} column`, + ); + recovered++; + } catch (err: unknown) { + const errorMessage = err instanceof Error ? err.message : String(err); + log.warn(`Failed to recover advanced triage task ${snapshot.id}: ${errorMessage}`); + } finally { + releaseReservation?.(); + } + } + return recovered; + } catch (err: unknown) { + const errorMessage = err instanceof Error ? err.message : String(err); + log.error(`Advanced triage recovery failed: ${errorMessage}`); + return 0; + } + } + /** * Clear stale transient merge statuses when no active merger owns the task. * diff --git a/packages/engine/src/triage.ts b/packages/engine/src/triage.ts index e780b4098d..bc67b306a2 100644 --- a/packages/engine/src/triage.ts +++ b/packages/engine/src/triage.ts @@ -215,6 +215,8 @@ export class TriageProcessor { /** The interval (ms) of the currently active `setInterval` timer. */ private activePollMs: number | null = null; private processing = new Set(); + /** Synchronous ownership fence shared with advanced-triage self-healing. */ + private advancedRecoveryReservations = new Set(); /** Timestamps when tasks entered the `processing` set, for staleness detection. */ private processingSince = new Map(); private wasGlobalPaused = false; @@ -647,6 +649,16 @@ export class TriageProcessor { * still completing Plan Review → todo after a stuck-kill of the main session. */ getProcessingTaskIds(): Set { + const ids = this.getPlanningTaskIds(); + for (const taskId of this.advancedRecoveryReservations) ids.add(taskId); + return ids; + } + + /** + * Return tasks owned by actual planner work, excluding recovery reservations. + * Recovery uses this narrower view when revalidating its own reserved task. + */ + getPlanningTaskIds(): Set { const ids = new Set(this.processing); for (const taskId of this.finalizing) ids.add(taskId); for (const taskId of this.activeSubagentSessions.keys()) { @@ -656,6 +668,23 @@ export class TriageProcessor { return ids; } + /** + * Reserve a task for advanced-state recovery unless planning already owns it. + * The check-and-add is synchronous, as is specifyTask's reciprocal guard, so + * neither side can slip between ownership inspection and acquisition. + */ + tryReserveAdvancedRecovery(taskId: string): (() => void) | undefined { + if ( + this.advancedRecoveryReservations.has(taskId) + || this.processing.has(taskId) + || this.hasLivePlanningWork(taskId) + ) { + return undefined; + } + this.advancedRecoveryReservations.add(taskId); + return () => this.advancedRecoveryReservations.delete(taskId); + } + /** True when this processor still owns live work for `taskId` (main, subagent, or finalize). */ private hasLivePlanningWork(taskId: string): boolean { if (this.finalizing.has(taskId)) return true; @@ -1042,7 +1071,9 @@ export class TriageProcessor { } const eligibleTriageTasks = allTasks.filter( - (t) => t.column === "triage" && !this.processing.has(t.id) && !this.hasLivePlanningWork(t.id) && !t.paused + (t) => t.column === "triage" && isTaskStillInPlanningStage(t) + && !this.advancedRecoveryReservations.has(t.id) + && !this.processing.has(t.id) && !this.hasLivePlanningWork(t.id) && !t.paused // Skip tasks awaiting manual plan approval — they should not be auto-discovered && t.status !== "awaiting-approval" // Skip failed specifications until the user explicitly retries them. @@ -1197,7 +1228,11 @@ export class TriageProcessor { `processing` was cleared by a stuck-kill eviction race. Concurrent claim is what leaves `status:"planning"` on a todo card after Plan Review APPROVE. */ - if (this.processing.has(task.id) || this.hasLivePlanningWork(task.id)) return; + if ( + this.advancedRecoveryReservations.has(task.id) + || this.processing.has(task.id) + || this.hasLivePlanningWork(task.id) + ) return; this.processing.add(task.id); this.processingSince.set(task.id, Date.now());