From 5ea98f7d4b6fb68243d44a3307cb3a4080eb882c Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Sun, 26 Jul 2026 17:33:35 -0700 Subject: [PATCH] fix: release admission claims when triage evicts a hung planner MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit evictStaleProcessing cleared `processing` but left the task in `coordinatorAdmittedTaskIds`, which is only cleared by specifyTask's finally — the path a hung promise never reaches. The card stayed eligible (so the throttle branch never logged or emitted `task:plan-admission-throttled`) while admitOldest's refresh filtered it out, leaving it on the "Queued to plan" badge with free slots and no diagnostic until engine restart. Also drop an untransferred pre-held host slot, which otherwise waits out the 600s stale-excess valve. Regression tests assert the invariant on the real production candidate source: an evicted card is re-offered and its host slot returned, while a still-live stale task keeps both claims. Co-Authored-By: Claude Opus 5 (1M context) --- .../triage-eviction-admission-claim-leak.md | 7 + packages/engine/src/__tests__/triage.test.ts | 121 ++++++++++++++++++ packages/engine/src/triage.ts | 26 ++++ 3 files changed, 154 insertions(+) create mode 100644 .changeset/triage-eviction-admission-claim-leak.md diff --git a/.changeset/triage-eviction-admission-claim-leak.md b/.changeset/triage-eviction-admission-claim-leak.md new file mode 100644 index 0000000000..a38e3fd24b --- /dev/null +++ b/.changeset/triage-eviction-admission-claim-leak.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Fix cards stuck on "Queued to plan" with free concurrency slots after a hung planner. +category: fix +dev: TriageProcessor.evictStaleProcessing now also clears `coordinatorAdmittedTaskIds` and drops any untransferred pre-held host slot, so an evicted planner's card is re-offered by the admission coordinator's refresh instead of being filtered out until engine restart. diff --git a/packages/engine/src/__tests__/triage.test.ts b/packages/engine/src/__tests__/triage.test.ts index fa2d95a19a..d7a25eb820 100644 --- a/packages/engine/src/__tests__/triage.test.ts +++ b/packages/engine/src/__tests__/triage.test.ts @@ -8,6 +8,13 @@ import { readAttachmentContents, computeUserCommentFingerprint, } from "../triage.js"; +import { + AgentSemaphore, + clearPreHeldExecutorSlotsForTests, + hasPreHeldExecutorSlot, + projectAdmissionCoordinator, + registerPreHeldExecutorSlot, +} from "../concurrency.js"; import { join } from "node:path"; import { readFileSync } from "node:fs"; import { mkdir, writeFile, rm, mkdtemp } from "node:fs/promises"; @@ -6763,6 +6770,120 @@ describe("evictStaleProcessing", () => { expect(processor.getProcessingTaskIds().has("FN-1312")).toBe(true); }); + /* + FNXC:ConcurrencyAdmission 2026-07-26-14:20: + Original symptom: a Todo card sat on the "Queued to plan" badge indefinitely with free + concurrency slots, and NEITHER diagnostic explained it — no "Plan throttled by …" log line and no + `task:plan-admission-throttled` run-audit row — because the card was still eligible (so the + throttle branch never ran) while `admitOldest`'s refresh filtered it out on a stale + `coordinatorAdmittedTaskIds` entry left behind by a hung planner promise that eviction reclaimed. + + Invariant under test (not just the reported repro): eviction releases EVERY admission-side claim + the evicted planner held — the coordinator admitted marker AND an untransferred pre-held host slot + — and does so on the real production candidate source, while a RETAINED (still-live) stale task + keeps both claims so eviction can never strip a running planner's capacity. + */ + describe("releases admission claims on eviction", () => { + const EVICT_ROOT = "/tmp/root-admission-claims"; + + /** The production candidate closure the coordinator actually calls (triage.ts constructor). */ + function providerRefresh(processor: TriageProcessor): () => Promise> { + const provider = (projectAdmissionCoordinator as any).providers.get(EVICT_ROOT)?.get(`specify:${EVICT_ROOT}`); + expect(provider).toBeDefined(); + return provider.refresh; + } + + function triageCard(id: string): Task { + return { + id, + title: "Hung planner", + description: "Test", + column: "triage", + dependencies: [], + steps: [], + currentStep: 0, + log: [], + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + } as Task; + } + + afterEach(() => { + clearPreHeldExecutorSlotsForTests(); + }); + + it("re-offers the evicted card to admission instead of hiding it forever", async () => { + const store = createMockStore({ listTasks: vi.fn().mockResolvedValue([triageCard("FN-HUNG")]) }); + const processor = new TriageProcessor(store, EVICT_ROOT); + (processor as any).running = true; + const refresh = providerRefresh(processor); + + vi.setSystemTime(new Date("2026-01-01T00:00:00.000Z")); + // Exactly the state a coordinator handoff leaves behind: admitted marker + processing claim, + // with a promise that never settles so specifyTask's finally never runs. + (processor as any).coordinatorAdmittedTaskIds.add("FN-HUNG"); + (processor as any).processing.add("FN-HUNG"); + (processor as any).processingSince.set("FN-HUNG", Date.now()); + + // While the claim is live the card must NOT be re-offered (no concurrent planner). + expect(await refresh()).toEqual([]); + + vi.setSystemTime(new Date("2026-01-01T00:31:00.000Z")); + expect(processor.evictStaleProcessing()).toEqual(new Set(["FN-HUNG"])); + + expect((processor as any).coordinatorAdmittedTaskIds.has("FN-HUNG")).toBe(false); + expect((await refresh()).map((candidate) => candidate.taskId)).toEqual(["FN-HUNG"]); + + (processor as any).unregisterAdmissionProvider?.(); + }); + + it("returns an untransferred pre-held host slot to the semaphore", () => { + const store = createMockStore(); + const semaphore = new AgentSemaphore(2); + const processor = new TriageProcessor(store, EVICT_ROOT, { semaphore }); + + vi.setSystemTime(new Date("2026-01-01T00:00:00.000Z")); + expect(semaphore.tryAcquire()).toBe(true); + registerPreHeldExecutorSlot("FN-HUNG"); + (processor as any).coordinatorAdmittedTaskIds.add("FN-HUNG"); + (processor as any).processing.add("FN-HUNG"); + (processor as any).processingSince.set("FN-HUNG", Date.now()); + expect(semaphore.activeCount).toBe(1); + + vi.setSystemTime(new Date("2026-01-01T00:31:00.000Z")); + expect(processor.evictStaleProcessing()).toEqual(new Set(["FN-HUNG"])); + + expect(hasPreHeldExecutorSlot("FN-HUNG")).toBe(false); + expect(semaphore.activeCount).toBe(0); + + (processor as any).unregisterAdmissionProvider?.(); + }); + + it("keeps both claims for a retained task whose planning session is still live", () => { + const store = createMockStore(); + const semaphore = new AgentSemaphore(2); + const processor = new TriageProcessor(store, EVICT_ROOT, { semaphore }); + + vi.setSystemTime(new Date("2026-01-01T00:00:00.000Z")); + expect(semaphore.tryAcquire()).toBe(true); + registerPreHeldExecutorSlot("FN-LIVE"); + (processor as any).coordinatorAdmittedTaskIds.add("FN-LIVE"); + (processor as any).processing.add("FN-LIVE"); + (processor as any).processingSince.set("FN-LIVE", Date.now()); + (processor as any).activeSessions.set("FN-LIVE", { dispose: vi.fn() }); + + vi.setSystemTime(new Date("2026-01-01T00:31:00.000Z")); + expect(processor.evictStaleProcessing()).toEqual(new Set()); + + expect((processor as any).coordinatorAdmittedTaskIds.has("FN-LIVE")).toBe(true); + expect(hasPreHeldExecutorSlot("FN-LIVE")).toBe(true); + expect(semaphore.activeCount).toBe(1); + + (processor as any).unregisterAdmissionProvider?.(); + semaphore.release(); + }); + }); + it("includes finalizing and subagent tasks in getProcessingTaskIds even when not in processing", () => { const store = createMockStore(); const processor = new TriageProcessor(store, "/tmp/root"); diff --git a/packages/engine/src/triage.ts b/packages/engine/src/triage.ts index e2136bbaf4..66b9f00a86 100644 --- a/packages/engine/src/triage.ts +++ b/packages/engine/src/triage.ts @@ -986,6 +986,32 @@ export class TriageProcessor { this.activeSessions.delete(taskId); this.stuckAborted.delete(taskId); this.finalizing.delete(taskId); + /* + FNXC:ConcurrencyAdmission 2026-07-26-14:20: + Eviction must release EVERY admission-side claim the hung planner still holds, not just + `processing`. Symptom this fixes: an operator reported a Todo card stuck on "Queued to plan" + with free concurrency slots and NO explanation in either diagnostic — no "Plan throttled by" + log line and no `task:plan-admission-throttled` run-audit row. + + Cause: `coordinatorAdmittedTaskIds` was only cleared by specifyTask's `finally` (and its + duplicate-claim guard), so a promise that never settles — exactly the case this eviction + exists for — left the id in the set permanently. Planning discovery does not consult that + set, so the card stayed in `triageTasks` and `maxToStart` stayed positive, which means the + throttle branch (the only thing that logs or emits) never fired; but `admitOldest`'s + `refresh()` filters on the set, so the coordinator saw no candidate. Silent stall until + engine restart, and the badge (a pure client-side "unplanned + idle in Todo" inference) kept + claiming the card was queued. + + The pre-held host slot is the second claim on the same path. A promise hung INSIDE + retryableWork has already transferred ownership, so the drop is a no-op there by design; a + promise hung BEFORE `takePreHeldExecutorSlot` still holds an untransferred registration, and + returning it here is the difference between a reclaimed slot and one the semaphore's + stale-excess valve cannot touch for 600s. If such a run later resumes, its take() returns + false and it acquires through `semaphore.run` normally, so the register/take-or-drop pairing + invariant holds either way. + */ + this.coordinatorAdmittedTaskIds.delete(taskId); + dropPreHeldExecutorSlot(taskId, this.options.semaphore); evicted.add(taskId); }