From 2d263acc49a2f1a0a53f97d8da9ea364db80f08e Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Sun, 26 Jul 2026 09:26:13 -0700 Subject: [PATCH] fix(FN-8600): keep self-healing from pausing live planners and unstick queued planning MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Planning moved into the task's own worktree but never published that path to activeSessionRegistry, so the self-owned-branch reclaim sweep's FN-4819 liveness guard was blind to a live planner. A zero-commit task branch trivially reads as tip-already-merged, so the sweep ran `git worktree remove --force` on the tree a planning session was using, the removal failed, and the failure escalated to branch-conflict-unrecoverable — parking a healthy card paused with no operator action. Planning now claims its worktree through acquireActiveSessionPath (new "planning" session kind) and releases it only while it still owns the record, so a live executor that took over the same path mid-teardown is never cleared. Also fixes planning starvation and its diagnosability: - admitOldest walks past candidates whose lane declines instead of ending the pass on candidates[0], unwinding each declined attempt's pre-held executor slot and reservation exactly so a decline cannot leak capacity past maxConcurrent. - Withheld planning admission emits a deduped task:plan-admission-throttled run-audit event (ids/counts only), written fire-and-forget with the dedupe marker set only after the write lands. Previously the binding gate lived only in a log line that is persisted nowhere, so "why did this card sit queued to plan?" was unanswerable after the fact. Reviewed by 8 review agents; every finding acted on or recorded. A proposed STALE_SEMAPHORE_EXCESS_REPAIR_MS 600s->180s reduction was reverted under review — nested runs are already excluded from the reclaim floor, so the window guards uncounted top-level holders such as a merge body, and shortening it would trade a bounded visible stall for an unbounded silent cap breach. Co-Authored-By: Claude Opus 5 (1M context) --- .changeset/planning-worktree-reclaim-race.md | 7 + AGENTS.md | 1 + .../engine/src/__tests__/concurrency.test.ts | 172 +++++++++++ ...iage-plan-admission-throttle-audit.test.ts | 236 +++++++++++++++ ...ning-worktree-session-registration.test.ts | 285 ++++++++++++++++++ .../engine/src/active-session-registry.ts | 15 +- packages/engine/src/concurrency.ts | 100 ++++-- packages/engine/src/run-audit.ts | 8 + packages/engine/src/triage.ts | 170 ++++++++++- 9 files changed, 974 insertions(+), 20 deletions(-) create mode 100644 .changeset/planning-worktree-reclaim-race.md create mode 100644 packages/engine/src/__tests__/triage-plan-admission-throttle-audit.test.ts create mode 100644 packages/engine/src/__tests__/triage-planning-worktree-session-registration.test.ts diff --git a/.changeset/planning-worktree-reclaim-race.md b/.changeset/planning-worktree-reclaim-race.md new file mode 100644 index 0000000000..b9f1ff4fc0 --- /dev/null +++ b/.changeset/planning-worktree-reclaim-race.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Stop self-healing pausing cards whose planning session is still running, and unstick queued planning. +category: fix +dev: Planning sessions now claim their worktree through `acquireActiveSessionPath` (new `"planning"` kind) and release it only when they still own the record, so the FN-4819 liveness guard in the self-owned-branch reclaim sweep defers instead of removing a live worktree and escalating to `branch-conflict-unrecoverable` — and planning's teardown cannot clear an executor entry that took over the same path. `ProjectAdmissionCoordinator.admitOldest` walks past candidates whose lane declines rather than ending the pass on `candidates[0]`, unwinding each declined attempt's pre-held executor slot and reservation. Withheld planning admission emits a deduped `task:plan-admission-throttled` run-audit event (ids/counts only), written fire-and-forget with the dedupe marker set only after the write lands. diff --git a/AGENTS.md b/AGENTS.md index 08b17b7f50..0804392af9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -285,6 +285,7 @@ Scoped exception (FN-5819): shared-branch-group members (`branchContext.assignme - FN-8004: `agent:heartbeat-move-skipped-soft-delete` records a heartbeat move that races a soft-deleted task without parking the durable agent. Metadata remains ids/timestamps/source only (`agentId`, optional `taskId`/`deletedAt`, `moveAttemptedAt`, optional `source`); it never stores error prose. - FN-8141: the executor's `fn_task_done(outcome="blocked", reason=..., blockedBy?=[...])` honest-blocked exit emits `task:execution-blocked-parked` when an executor parks a genuinely-impossible task `failed` (`error = "BLOCKED: "`) instead of laundering it to `done` by skipping steps. It bypasses the completion/verdict/bulk-completion gates (blocked is not a completion claim), leaves steps in their true statuses, preserves worktree/branch, records `blockedBy` as real `task.dependencies` edges so the task requeues behind the blocker, and does NOT hand off to review — the parked row is honored by the executor's `status === "failed"` post-loop branch and is not auto-recovered into in-review by `recoverStrandedCompletedTodoTasks` (steps are not all done/skipped and `task.error` is set). Metadata stays ids/outcomes-only (`taskId`, `blockedBy` ids, `hasReason` boolean — never the reason prose). - FN-8305: durable symbol-lock operations emit `symbol-lock:acquired`, `symbol-lock:acquire-conflict`, `symbol-lock:renewed`, `symbol-lock:released`, `symbol-lock:reconcile-stale`, and deduplicated `symbol-lock:reconcile-stale-no-action`. Metadata is ids/counts/outcomes-only; normalized opaque symbol keys are permitted IDs, while raw symbol prose is not. +- FN-8600: triage emits `task:plan-admission-throttled` when planning admission is withheld while eligible cards are waiting, recording the binding gate (`blockedBy`: `"running-agent cap"` or `"global semaphore"`) plus `maxConcurrent`, `claimed`, `projectRoom`, `eligibleCount`, up to five `eligibleTaskIds`, `processingCount`, up to five `processingTaskIds`, and the semaphore `activeCount`/`limit`/`availableCount`/`waitingCount`. Metadata is ids/counts-only. Deduped on the gate signature INCLUDING the eligible task IDs, so a sustained stall collapses to one row while a new card's stall is never swallowed; the marker is set only after the write lands, so a failed write retries on the next poll. Purpose: before this event the binding gate existed only in a `planLog` line that is persisted nowhere, so "why did this card sit queued to plan?" was unanswerable after the fact. Reachable today by direct DB query only — the sole run-audit read route resolves through a durable agent's heartbeat run and this event uses a synthetic run id under `agentId:"triage"`. - FN-8592: startup and periodic self-healing emit `task:reconcile-stranded-hold-continuation` when an idle hold-column card with a real spec is re-seeded at its pre-release Plan Review, and deduped `task:reconcile-stranded-hold-continuation-no-action` for a candidate guard or race loss. Metadata stays ids/counts/outcomes-only (`taskId`, `column`, node/workflow identifiers, staleness or reason); healthy non-candidates are silent. The repair is insert-only and uses the shared per-task advisory transaction lock; global/engine pause and `autoMerge:false` defer to the operator. - FN-8492: the self-healing sweep `reconcile-orphaned-pending-step-results` (startup, right after legacy adoption, plus periodic maintenance) emits `task:reconcile-orphaned-pending-step-results` when it REWRITES `pending` workflow-step results with no live session behind them to `failed` (canonical liveness triple: `activeSessionRegistry` path, `executingTaskLock`, `isTaskActive`). It must never DELETE an orphaned entry — the merge gate blocks on pending/failed results, not on an enabled step with no result, so deletion silently satisfies the gate and the task merges with its review skipped; the `failed` rewrite keeps the gate closed and hands re-run/bypass to the failed-pre-merge-steps recovery and FN-7720 operator-bypass paths. `in-progress` rows are always skipped (executor-owned; resume is deferred at startup), the row is re-read immediately before the write, and user pauses are never disturbed. Metadata is ids/counts-only (`taskId`, `column`, `orphanedCount`, `resultCount`). - FN-8356: self-healing emits `task:reconcile-stale-duplicate-decision` when it clears a triage-marker duplicate-decision pause against a missing, deleted, done, or archived canonical. Metadata is ids/outcomes-only (`taskId`, `canonicalId`, `canonicalColumn`, `canonicalDeleted`, `priorPausedReason`); active canonical decisions and user pauses remain untouched. diff --git a/packages/engine/src/__tests__/concurrency.test.ts b/packages/engine/src/__tests__/concurrency.test.ts index aa96383a1a..c844003796 100644 --- a/packages/engine/src/__tests__/concurrency.test.ts +++ b/packages/engine/src/__tests__/concurrency.test.ts @@ -1083,6 +1083,178 @@ describe("ProjectAdmissionCoordinator", () => { expect(started).toEqual(["old", "other-project"]); }); + /* + FNXC:ConcurrencyAdmission 2026-07-26-09:45: + Regression for the planning-starvation half of the FN-8600 incident: a card sat "Queued to plan" + while capacity was free, because admitOldest only ever evaluated candidates[0]. When the oldest + candidate's lane declines the handoff, younger work in another lane must still be admitted. + + Invariant under test (not just the reported repro): a declining candidate is SKIPPED, not + vetoing; age order is still respected among the candidates that can start; the declined + candidate's reservation and host slot are fully returned; and a single call still admits at most + one task. + */ + it("skips candidates whose lane declines and admits the next oldest instead", async () => { + const coordinator = new ProjectAdmissionCoordinator(); + const semaphore = new AgentSemaphore(4); + const started: string[] = []; + + const admitted = await coordinator.admitOldest({ + projectId: "project-a", + maxConcurrent: 4, + claimed: () => 0, + semaphore, + refresh: async () => [ + // Oldest, but its lane cannot start it (e.g. a merge id no longer queued). + { + taskId: "FN-OLDEST", projectId: "project-a", createdAt: "2026-01-01T00:00:00.000Z", + start: async () => { started.push("FN-OLDEST"); return false; }, + }, + // Also declines — proves the walk continues past more than one. + { + taskId: "FN-MIDDLE", projectId: "project-a", createdAt: "2026-01-02T00:00:00.000Z", + start: async () => { started.push("FN-MIDDLE"); return false; }, + }, + // The planning candidate that was starving behind them. + { + taskId: "FN-PLANNING", projectId: "project-a", createdAt: "2026-01-03T00:00:00.000Z", + start: async () => { started.push("FN-PLANNING"); }, + }, + // Younger still: must NOT be admitted, so skipping never becomes overtaking. + { + taskId: "FN-YOUNGEST", projectId: "project-a", createdAt: "2026-01-04T00:00:00.000Z", + start: async () => { started.push("FN-YOUNGEST"); }, + }, + ], + }); + + expect(admitted).toBe("FN-PLANNING"); + expect(started).toEqual(["FN-OLDEST", "FN-MIDDLE", "FN-PLANNING"]); + // Exactly one host slot is held — by the admitted task, not the decliners. + expect(semaphore.activeCount).toBe(1); + + coordinator.releaseReservation("FN-PLANNING"); + semaphore.release(); + expect(semaphore.activeCount).toBe(0); + }); + + /* + FNXC:ConcurrencyAdmission 2026-07-26-10:35: + Compatibility shims have no `tryAcquire`, so they never take a host slot. Releasing one anyway + returns capacity nobody held — `returnSlot` decrements `_active` and drains a waiter regardless. + Walking past decliners would repeat that once per decliner, so N decliners could free N phantom + slots and push concurrency past maxConcurrent. + */ + it("does not release phantom host slots when the semaphore shim cannot reserve", async () => { + const coordinator = new ProjectAdmissionCoordinator(); + const releases: number[] = []; + // A pre-tryAcquire shim: release only, no tryAcquire. + const shim = { release: () => { releases.push(1); } }; + + const admitted = await coordinator.admitOldest({ + projectId: "project-shim", + maxConcurrent: 4, + claimed: () => 0, + semaphore: shim as unknown as Parameters[0]["semaphore"], + refresh: async () => [ + { taskId: "FN-A", projectId: "project-shim", createdAt: "2026-01-01T00:00:00.000Z", start: async () => false }, + { taskId: "FN-B", projectId: "project-shim", createdAt: "2026-01-02T00:00:00.000Z", start: async () => false }, + { taskId: "FN-C", projectId: "project-shim", createdAt: "2026-01-03T00:00:00.000Z", start: async () => undefined }, + ], + }); + + expect(admitted).toBe("FN-C"); + // Two decliners must not have produced two unmatched releases. + expect(releases).toHaveLength(0); + }); + + /* + FNXC:ConcurrencyAdmission 2026-07-26-11:05: + The triage and scheduler lanes set `reserve: () => registerPreHeldExecutorSlot(id)`. If a decline + unwinds the semaphore but not that registration, the id stays in the module-global pre-held set + with no backing acquire — the next pass's `takePreHeldExecutorSlot` then runs a full top-level + session without acquiring a slot and releases one it never held, leaving `_active` permanently + below the live agent count and the global cap silently breached. + */ + it("drops a declined candidate's pre-held executor slot, not just the semaphore slot", async () => { + clearPreHeldExecutorSlotsForTests(); + const coordinator = new ProjectAdmissionCoordinator(); + const semaphore = new AgentSemaphore(4); + + const admitted = await coordinator.admitOldest({ + projectId: "project-prehold", + maxConcurrent: 4, + claimed: () => 0, + semaphore, + refresh: async () => [ + { + taskId: "FN-DECLINE", projectId: "project-prehold", createdAt: "2026-01-01T00:00:00.000Z", + reserve: () => registerPreHeldExecutorSlot("FN-DECLINE"), + start: async () => false, + }, + { + taskId: "FN-TAKES", projectId: "project-prehold", createdAt: "2026-01-02T00:00:00.000Z", + reserve: () => registerPreHeldExecutorSlot("FN-TAKES"), + start: async () => undefined, + }, + ], + }); + + expect(admitted).toBe("FN-TAKES"); + // The decliner must leave nothing behind that a later pass could "take". + expect(hasPreHeldExecutorSlot("FN-DECLINE")).toBe(false); + // Only the admitted candidate still holds a slot. + expect(semaphore.activeCount).toBe(1); + expect(hasPreHeldExecutorSlot("FN-TAKES")).toBe(true); + + dropPreHeldExecutorSlot("FN-TAKES", semaphore); + expect(semaphore.activeCount).toBe(0); + clearPreHeldExecutorSlotsForTests(); + }); + + it("returns the reservation and host slot when a candidate's start() throws", async () => { + const coordinator = new ProjectAdmissionCoordinator(); + const semaphore = new AgentSemaphore(2); + + await expect(coordinator.admitOldest({ + projectId: "project-throw", + maxConcurrent: 4, + claimed: () => 0, + semaphore, + refresh: async () => [{ + taskId: "FN-BOOM", projectId: "project-throw", createdAt: "2026-01-01T00:00:00.000Z", + start: async () => { throw new Error("lane exploded"); }, + }], + })).rejects.toThrow("lane exploded"); + + // A thrown lane must not strand capacity. + expect(semaphore.activeCount).toBe(0); + }); + + it("stops the walk when the host semaphore is exhausted rather than spinning candidates", async () => { + const coordinator = new ProjectAdmissionCoordinator(); + const semaphore = new AgentSemaphore(1); + // Exhaust the host semaphore so no candidate can acquire a slot. + expect(semaphore.tryAcquire()).toBe(true); + const started: string[] = []; + + const admitted = await coordinator.admitOldest({ + projectId: "project-a", + maxConcurrent: 4, + claimed: () => 0, + semaphore, + refresh: async () => [ + { taskId: "FN-1", projectId: "project-a", createdAt: "2026-01-01T00:00:00.000Z", start: async () => { started.push("FN-1"); } }, + { taskId: "FN-2", projectId: "project-a", createdAt: "2026-01-02T00:00:00.000Z", start: async () => { started.push("FN-2"); } }, + ], + }); + + expect(admitted).toBeUndefined(); + expect(started).toEqual([]); + expect(semaphore.activeCount).toBe(1); + semaphore.release(); + }); + it("releases a rejected handoff and retains an accepted reservation until lane transfer", async () => { const coordinator = new ProjectAdmissionCoordinator(); const semaphore = new AgentSemaphore(1); diff --git a/packages/engine/src/__tests__/triage-plan-admission-throttle-audit.test.ts b/packages/engine/src/__tests__/triage-plan-admission-throttle-audit.test.ts new file mode 100644 index 0000000000..5ca1961a39 --- /dev/null +++ b/packages/engine/src/__tests__/triage-plan-admission-throttle-audit.test.ts @@ -0,0 +1,236 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import type { Settings, Task, TaskStore } from "@fusion/core"; +import { AgentSemaphore } from "../concurrency.js"; +import { TriageProcessor } from "../triage.js"; + +/* +FNXC:ConcurrencyAdmission 2026-07-26-09:30: +Regression suite for the diagnosability half of the FN-8600 incident. + +Original symptom: an operator asked why a started card sat "Queued to plan" for seven minutes. The +answer was unrecoverable after the fact — the binding gate was written ONLY to `planLog.log`, which +lands in the TUI's in-memory pane (truncated) and is persisted nowhere. Reconstructing the timeline +required a manual DB forensics pass that still could not separate "host semaphore exhausted" from +"project cap consumed". + +Invariant under test: whenever planning admission is withheld while eligible work exists, the +binding gate is recorded durably in run-audit with ids/counts only — and a sustained stall records +ONE row rather than one per poll, so the signal stays readable. +*/ + +vi.mock("../reviewer.js", () => ({ reviewStep: vi.fn() })); + +vi.mock("../pi.js", () => { + class ModelFallbackExhaustedError extends Error {} + return { + ModelFallbackExhaustedError, + createFnAgent: vi.fn(), + describeModel: vi.fn().mockReturnValue("mock-model"), + formatModelMarkerDetails: vi.fn((model: string) => model), + promptWithFallback: vi.fn().mockResolvedValue(undefined), + }; +}); + +vi.mock("@fusion/core", async (importOriginal) => { + const { createEngineCoreMock } = await import("../test/mockCore.js"); + const original = await importOriginal(); + return createEngineCoreMock(() => Promise.resolve(original)); +}); + +/** A todo card with no steps and no spec on disk — i.e. eligible for planning. */ +function eligibleTodoTask(id: string): Task { + return { + id, + description: "Add ability to favorite projects on mobile", + column: "todo", + dependencies: [], + steps: [], + currentStep: 0, + log: [], + createdAt: "2026-07-26T15:37:52.786Z", + updatedAt: "2026-07-26T15:37:52.786Z", + } as Task; +} + +interface RecordedEvent { type: string; target: string; metadata?: Record } + +function createStore(tasks: Task[], recorded: RecordedEvent[], settings: Partial = {}): TaskStore { + return { + getTask: vi.fn().mockImplementation(async (id: string) => { + const task = tasks.find((candidate) => candidate.id === id); + return task ? { ...task, prompt: "", attachments: [], comments: [] } : null; + }), + listTasks: vi.fn().mockResolvedValue(tasks), + getSettings: vi.fn().mockResolvedValue({ + maxConcurrent: 12, + maxWorktrees: 4, + pollIntervalMs: 600_000, + groupOverlappingFiles: false, + autoMerge: true, + ...settings, + } as Settings), + recordRunAuditEvent: vi.fn().mockImplementation(async (event: { mutationType: string; target: string; metadata?: Record }) => { + recorded.push({ type: event.mutationType, target: event.target, metadata: event.metadata }); + }), + updateTask: vi.fn().mockResolvedValue(undefined), + logEntry: vi.fn().mockResolvedValue(undefined), + appendAgentLog: vi.fn().mockResolvedValue(undefined), + moveTask: vi.fn(), + createTask: vi.fn(), + deleteTask: vi.fn(), + mergeTask: vi.fn(), + updateSettings: vi.fn(), + getAgentLogs: vi.fn().mockResolvedValue([]), + addSteeringComment: vi.fn(), + parseDependenciesFromPrompt: vi.fn().mockResolvedValue([]), + parseStepsFromPrompt: vi.fn().mockResolvedValue([]), + parseFileScopeFromPrompt: vi.fn().mockResolvedValue([]), + getTaskWorkflowSelection: vi.fn().mockReturnValue(undefined), + getWorkflowDefinition: vi.fn().mockResolvedValue(undefined), + on: vi.fn(), + emit: vi.fn(), + } as unknown as TaskStore; +} + +/** Runs one real poll pass against an exhausted host semaphore. */ +async function pollWithExhaustedSemaphore(store: TaskStore, semaphore: AgentSemaphore): Promise { + const processor = new TriageProcessor(store, "/tmp/fn-8600-throttle-root", { semaphore }); + // poll() is a no-op unless the processor is running; these tests drive one pass directly rather + // than starting the interval timer, which would make them time-dependent. + (processor as unknown as { running: boolean }).running = true; + await (processor as unknown as { poll: () => Promise }).poll(); + // The audit write is fire-and-forget, so let its microtask settle before asserting. + await new Promise((resolve) => setImmediate(resolve)); + return processor; +} + +describe("plan admission throttle run-audit (FN-8600)", () => { + let recorded: RecordedEvent[]; + + beforeEach(() => { + vi.clearAllMocks(); + recorded = []; + }); + + it("records the binding gate when planning is withheld with eligible work", async () => { + const semaphore = new AgentSemaphore(1); + expect(semaphore.tryAcquire()).toBe(true); // host capacity fully spent + + const store = createStore([eligibleTodoTask("FN-8600")], recorded); + await pollWithExhaustedSemaphore(store, semaphore); + + const throttle = recorded.filter((event) => event.type === "task:plan-admission-throttled"); + expect(throttle).toHaveLength(1); + // The gate the operator could not previously determine. + expect(throttle[0].metadata).toMatchObject({ + blockedBy: "global semaphore", + maxConcurrent: 12, + eligibleCount: 1, + eligibleTaskIds: ["FN-8600"], + semaphoreLimit: 1, + semaphoreAvailableCount: 0, + }); + semaphore.release(); + }); + + it("emits one row for a sustained stall instead of one per poll", async () => { + const semaphore = new AgentSemaphore(1); + expect(semaphore.tryAcquire()).toBe(true); + + const store = createStore([eligibleTodoTask("FN-8600")], recorded); + const processor = new TriageProcessor(store, "/tmp/fn-8600-throttle-root", { semaphore }); + (processor as unknown as { running: boolean }).running = true; + const poll = (processor as unknown as { poll: () => Promise }).poll.bind(processor); + await poll(); + await poll(); + await poll(); + await new Promise((resolve) => setImmediate(resolve)); + + expect(recorded.filter((event) => event.type === "task:plan-admission-throttled")).toHaveLength(1); + semaphore.release(); + }); + + /* + FNXC:ConcurrencyAdmission 2026-07-26-10:45: + A counts-only signature would swallow a NEW card's stall whenever the numbers happened to land on + the same tuple, leaving the only persisted row naming a different task — fatal for an event whose + whole job is answering "why is THIS card queued". + */ + it("emits again when a different card is the one stalling, even with identical counts", async () => { + const semaphore = new AgentSemaphore(1); + expect(semaphore.tryAcquire()).toBe(true); + + const first = eligibleTodoTask("FN-8600"); + const store = createStore([first], recorded); + const processor = new TriageProcessor(store, "/tmp/fn-8600-throttle-root", { semaphore }); + (processor as unknown as { running: boolean }).running = true; + const poll = (processor as unknown as { poll: () => Promise }).poll.bind(processor); + await poll(); + + // Same counts, different card: A leaves the queue as C enters. + (store.listTasks as unknown as { mockResolvedValue: (v: Task[]) => void }) + .mockResolvedValue([eligibleTodoTask("FN-8601")]); + await poll(); + + const throttle = recorded.filter((event) => event.type === "task:plan-admission-throttled"); + expect(throttle).toHaveLength(2); + expect(throttle[1].metadata).toMatchObject({ eligibleTaskIds: ["FN-8601"] }); + semaphore.release(); + }); + + /* + FNXC:ConcurrencyAdmission 2026-07-26-10:45: + The dedupe marker must not be poisoned by a write that never landed. Store contention is the exact + condition this event explains, so swallowing the record on write failure would reproduce the + original unanswerable-stall problem. + */ + it("retries the audit write on the next poll when the first write fails", async () => { + const semaphore = new AgentSemaphore(1); + expect(semaphore.tryAcquire()).toBe(true); + + const store = createStore([eligibleTodoTask("FN-8600")], recorded); + let attempts = 0; + (store as unknown as { recordRunAuditEvent: unknown }).recordRunAuditEvent = vi.fn() + .mockImplementation(async (event: { mutationType: string; target: string; metadata?: Record }) => { + attempts += 1; + if (attempts === 1) throw new Error("store contended"); + recorded.push({ type: event.mutationType, target: event.target, metadata: event.metadata }); + }); + + const processor = new TriageProcessor(store, "/tmp/fn-8600-throttle-root", { semaphore }); + (processor as unknown as { running: boolean }).running = true; + const poll = (processor as unknown as { poll: () => Promise }).poll.bind(processor); + + await poll(); + // Let the fire-and-forget write settle its rejection before the next pass. + await new Promise((resolve) => setImmediate(resolve)); + expect(recorded.filter((event) => event.type === "task:plan-admission-throttled")).toHaveLength(0); + + await poll(); + await new Promise((resolve) => setImmediate(resolve)); + expect(recorded.filter((event) => event.type === "task:plan-admission-throttled")).toHaveLength(1); + semaphore.release(); + }); + + it("stays silent when planning has capacity", async () => { + const semaphore = new AgentSemaphore(4); + const store = createStore([], recorded); + await pollWithExhaustedSemaphore(store, semaphore); + + expect(recorded.filter((event) => event.type === "task:plan-admission-throttled")).toHaveLength(0); + }); + + it("carries no prompt, title, or reason prose — ids and counts only", async () => { + const semaphore = new AgentSemaphore(1); + expect(semaphore.tryAcquire()).toBe(true); + + const store = createStore([eligibleTodoTask("FN-8600")], recorded); + await pollWithExhaustedSemaphore(store, semaphore); + + const throttle = recorded.find((event) => event.type === "task:plan-admission-throttled"); + expect(throttle).toBeDefined(); + const serialized = JSON.stringify(throttle!.metadata ?? {}); + expect(serialized).not.toContain("favorite projects"); + semaphore.release(); + }); +}); diff --git a/packages/engine/src/__tests__/triage-planning-worktree-session-registration.test.ts b/packages/engine/src/__tests__/triage-planning-worktree-session-registration.test.ts new file mode 100644 index 0000000000..b21bd22e03 --- /dev/null +++ b/packages/engine/src/__tests__/triage-planning-worktree-session-registration.test.ts @@ -0,0 +1,285 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import type { Settings, Task, TaskDetail, TaskStore } from "@fusion/core"; +import { activeSessionRegistry } from "../active-session-registry.js"; +import { TriageProcessor } from "../triage.js"; + +/* +FNXC:NodeWorktreeIsolation 2026-07-26-09:10: +Regression suite for the FN-8600 spurious-pause incident. + +Original symptom: a healthy card was parked `paused` with `pausedReason:"branch-conflict-unrecoverable"` +and `error: "Task branch conflict: fusion/fn-8600 is not safely reclaimable (tip-already-merged +cleanup failed for FN-8600)"` — with no operator action (`userPaused:false`). Cause: planning moved +into the task's own worktree (FNXC:NodeWorktreeIsolation 2026-07-25-22:10) but never registered that +path in `activeSessionRegistry`. A zero-commit task branch trivially reads as `tip-already-merged` +(its tip IS the integration ref), so the self-owned-branch reclaim sweep selected it, and its +FN-4819 liveness guard — `activeSessionRegistry.isPathActive(task.worktree)` — could not see the +live planner. `git worktree remove --force` then failed against the in-use worktree and the outer +catch escalated to `branch-conflict-unrecoverable`. + +Surface enumeration (the invariant is "a planner holding a worktree is a visible live session", +not merely "FN-8600 stops pausing"): + - registers the acquired worktree for the life of the planning run, with kind "planning" and the + owning task id, so EVERY isPathActive-keyed guard sees it — not just the reclaim sweep; + - releases it on the success path; + - releases it when planning throws, since a leaked entry would permanently veto legitimate cleanup + of that worktree (the inverse stall); + - does NOT register when planning falls back to the shared root checkout, which is the operator's + tree and must never be marked task-owned. +*/ + +const { mockCreateFnAgent, mockPromptWithFallback } = vi.hoisted(() => ({ + mockCreateFnAgent: vi.fn(), + mockPromptWithFallback: vi.fn(), +})); + +vi.mock("../reviewer.js", () => ({ + reviewStep: vi.fn(), +})); + +vi.mock("../pi.js", () => { + class ModelFallbackExhaustedError extends Error {} + return { + ModelFallbackExhaustedError, + createFnAgent: mockCreateFnAgent, + describeModel: vi.fn().mockReturnValue("mock-model"), + formatModelMarkerDetails: vi.fn((model: string) => model), + promptWithFallback: mockPromptWithFallback, + }; +}); + +vi.mock("@fusion/core", async (importOriginal) => { + const { createEngineCoreMock } = await import("../test/mockCore.js"); + const original = await importOriginal(); + return createEngineCoreMock(() => Promise.resolve(original)); +}); + +const ROOT_DIR = "/tmp/fn-8600-root"; +const PLANNING_WORKTREE = "/tmp/fn-8600-root/.worktrees/amber-stone"; + +function createTask(overrides: Partial = {}): Task { + return { + id: "FN-8600", + description: "Add ability to favorite projects on mobile", + column: "todo", + dependencies: [], + steps: [], + currentStep: 0, + log: [], + createdAt: "2026-07-26T15:37:52.786Z", + updatedAt: "2026-07-26T15:37:52.786Z", + ...overrides, + }; +} + +function createStore(task: Task): TaskStore { + return { + getTask: vi.fn().mockResolvedValue({ ...task, prompt: "", attachments: [], comments: [] } as TaskDetail), + listTasks: vi.fn().mockResolvedValue([]), + createTask: vi.fn(), + moveTask: vi.fn(), + updateTask: vi.fn().mockResolvedValue(undefined), + deleteTask: vi.fn(), + mergeTask: vi.fn(), + getSettings: vi.fn().mockResolvedValue({ + maxConcurrent: 12, + maxWorktrees: 4, + pollIntervalMs: 10000, + groupOverlappingFiles: false, + autoMerge: true, + } as Settings), + updateSettings: vi.fn(), + logEntry: vi.fn().mockResolvedValue(undefined), + appendAgentLog: vi.fn().mockResolvedValue(undefined), + getAgentLogs: vi.fn().mockResolvedValue([]), + addSteeringComment: vi.fn(), + parseDependenciesFromPrompt: vi.fn().mockResolvedValue([]), + parseStepsFromPrompt: vi.fn().mockResolvedValue([]), + parseFileScopeFromPrompt: vi.fn().mockResolvedValue([]), + getTaskWorkflowSelection: vi.fn().mockReturnValue(undefined), + getWorkflowDefinition: vi.fn().mockResolvedValue(undefined), + on: vi.fn(), + emit: vi.fn(), + } as unknown as TaskStore; +} + +/** + * Runs a planning session, invoking `duringSession` while the agent session is live. + * The observation point is `promptWithFallback` — triage drives the planner through it rather than + * calling `session.prompt` directly, so that is where "the session is running" is true. + */ +function stubAgentSession(duringSession: () => void, opts: { throwInSession?: boolean } = {}): void { + mockCreateFnAgent.mockImplementationOnce(async () => ({ + session: { + state: {}, + sessionManager: { getLeafId: vi.fn().mockReturnValue(null) }, + prompt: vi.fn().mockResolvedValue(undefined), + dispose: vi.fn(), + navigateTree: vi.fn(), + }, + })); + mockPromptWithFallback.mockImplementationOnce(async () => { + duringSession(); + if (opts.throwInSession) throw new Error("planner blew up"); + }); +} + +describe("planning session worktree registration (FN-8600)", () => { + beforeEach(() => { + vi.clearAllMocks(); + activeSessionRegistry.unregisterPath(PLANNING_WORKTREE); + activeSessionRegistry.unregisterPath(ROOT_DIR); + }); + + it("marks the planning worktree active for the life of the session, then releases it", async () => { + const task = createTask(); + const store = createStore(task); + let activeDuringSession: boolean | undefined; + let recordDuringSession: { taskId: string; kind: string } | undefined; + + stubAgentSession(() => { + activeDuringSession = activeSessionRegistry.isPathActive(PLANNING_WORKTREE); + const record = activeSessionRegistry.lookupByPath(PLANNING_WORKTREE); + recordDuringSession = record ? { taskId: record.taskId, kind: record.kind } : undefined; + }); + + await new TriageProcessor(store, ROOT_DIR, { + acquirePlanningWorktree: async () => PLANNING_WORKTREE, + }).specifyTask(task); + + // The exact condition the self-healing reclaim sweep consults before removing a worktree. + expect(activeDuringSession).toBe(true); + expect(recordDuringSession).toEqual({ taskId: "FN-8600", kind: "planning" }); + // And it must not outlive the run, or it would veto all later cleanup of that path. + expect(activeSessionRegistry.isPathActive(PLANNING_WORKTREE)).toBe(false); + }); + + it("releases the planning worktree when the planning session fails", async () => { + const task = createTask({ id: "FN-8600-FAIL" }); + const store = createStore(task); + let activeDuringSession: boolean | undefined; + + stubAgentSession(() => { + activeDuringSession = activeSessionRegistry.isPathActive(PLANNING_WORKTREE); + }, { throwInSession: true }); + + await new TriageProcessor(store, ROOT_DIR, { + acquirePlanningWorktree: async () => PLANNING_WORKTREE, + }).specifyTask(task); + + expect(activeDuringSession).toBe(true); + expect(activeSessionRegistry.isPathActive(PLANNING_WORKTREE)).toBe(false); + }); + + /* + FNXC:NodeWorktreeIsolation 2026-07-26-10:20: + The teardown must not clear a LIVE executor entry. `finalizeApprovedTask` moves the card to `todo` + while planning is still unwinding, so the scheduler can dispatch and the executor can register the + same worktree path (same task => `registerPath` overwrites) before planning's `finally` runs. A + bare path delete there would make `isPathActive` false under a live executor and hand the reclaim + sweep the same worktree it tore out from under a planner in FN-8600 — the fix reintroducing its + own symptom one lane over. + */ + it("leaves a live executor registration on the same path intact when planning unwinds", async () => { + const task = createTask({ id: "FN-8600-HANDOFF" }); + const store = createStore(task); + + stubAgentSession(() => { + // Simulate the planning->execution handoff landing mid-teardown: the executor takes over the + // very same worktree path for the very same task. + activeSessionRegistry.registerPath(PLANNING_WORKTREE, { + taskId: "FN-8600-HANDOFF", + kind: "executor", + ownerKey: "FN-8600-HANDOFF", + }); + }); + + await new TriageProcessor(store, ROOT_DIR, { + acquirePlanningWorktree: async () => PLANNING_WORKTREE, + }).specifyTask(task); + + // The executor's session must survive planning's finally, or the reclaim sweep removes its tree. + expect(activeSessionRegistry.isPathActive(PLANNING_WORKTREE)).toBe(true); + expect(activeSessionRegistry.lookupByPath(PLANNING_WORKTREE)?.kind).toBe("executor"); + activeSessionRegistry.unregisterPath(PLANNING_WORKTREE); + }); + + /* + FNXC:NodeWorktreeIsolation 2026-07-26-10:15: + The consuming guard is the point of the fix. `SelfHealingManager`'s self-owned-branch reclaim sweep + defers on the kind-agnostic predicate `activeSessionRegistry.isPathActive(task.worktree)` + (self-healing.ts, "deferring reclaim for"). Pinning that predicate against the "planning" kind is + what proves a live planner is now visible to it — and to the other liveness-keyed sweeps that share + the same predicate — rather than only proving that triage wrote a record. + */ + it("makes a live planner satisfy the kind-agnostic liveness predicate the reclaim sweep defers on", async () => { + const task = createTask({ id: "FN-8600-GUARD" }); + const store = createStore(task); + let guardDuringSession: boolean | undefined; + + stubAgentSession(() => { + guardDuringSession = activeSessionRegistry.isPathActive(PLANNING_WORKTREE); + }); + + await new TriageProcessor(store, ROOT_DIR, { + acquirePlanningWorktree: async () => PLANNING_WORKTREE, + }).specifyTask(task); + + expect(guardDuringSession).toBe(true); + expect(activeSessionRegistry.isPathActive(PLANNING_WORKTREE)).toBe(false); + }); + + it("falls back to the shared checkout when a live foreign task holds the planning worktree", async () => { + const task = createTask({ id: "FN-8600-CONTENDED" }); + const store = createStore(task); + // A different, live task already owns the path. + activeSessionRegistry.registerPath(PLANNING_WORKTREE, { + taskId: "FN-OTHER", + kind: "executor", + ownerKey: "FN-OTHER", + }); + let cwdDuringSession: string | undefined; + + mockCreateFnAgent.mockImplementationOnce(async (opts: { cwd?: string }) => { + cwdDuringSession = opts.cwd; + return { + session: { + state: {}, + sessionManager: { getLeafId: vi.fn().mockReturnValue(null) }, + prompt: vi.fn().mockResolvedValue(undefined), + dispose: vi.fn(), + navigateTree: vi.fn(), + }, + }; + }); + mockPromptWithFallback.mockImplementationOnce(async () => undefined); + + await new TriageProcessor(store, ROOT_DIR, { + acquirePlanningWorktree: async () => PLANNING_WORKTREE, + }).specifyTask(task); + + // Planning must not run inside a worktree a live foreign session owns... + expect(cwdDuringSession).toBe(ROOT_DIR); + // ...and must not have disturbed that session's registration on the way out. + expect(activeSessionRegistry.lookupByPath(PLANNING_WORKTREE)?.taskId).toBe("FN-OTHER"); + activeSessionRegistry.unregisterPath(PLANNING_WORKTREE); + }); + + it("never marks the shared root checkout as a task-owned session", async () => { + const task = createTask({ id: "FN-8600-SHARED" }); + const store = createStore(task); + let rootActiveDuringSession: boolean | undefined; + + stubAgentSession(() => { + rootActiveDuringSession = activeSessionRegistry.isPathActive(ROOT_DIR); + }); + + // No worktree available: planning falls back to the operator's checkout. + await new TriageProcessor(store, ROOT_DIR, { + acquirePlanningWorktree: async () => null, + }).specifyTask(task); + + expect(rootActiveDuringSession).toBe(false); + expect(activeSessionRegistry.isPathActive(ROOT_DIR)).toBe(false); + }); +}); diff --git a/packages/engine/src/active-session-registry.ts b/packages/engine/src/active-session-registry.ts index 7de9ab1f53..8b873b48d4 100644 --- a/packages/engine/src/active-session-registry.ts +++ b/packages/engine/src/active-session-registry.ts @@ -20,7 +20,20 @@ phase) and the land lease (merge phase) never overlap in time on the same path, keeping them distinct kinds (each released in its own `finally`) means a stale entry of one kind can never be mistaken for a live hold of the other. */ -export type ActiveSessionKind = "executor" | "step-session" | "workflow-step" | "step-session-parallel" | "ai-merge" | "workspace-repo-acquire" | "workspace-repo-land"; +/* +FNXC:NodeWorktreeIsolation 2026-07-26-09:10: +"planning" exists because FNXC:NodeWorktreeIsolation 2026-07-25-22:10 moved the triage/planning +session out of the shared checkout and into the TASK's own worktree, but planning never registered +that path here. Every liveness guard keyed on this registry — above all the FN-4819 skip in +`SelfHealingManager`'s self-owned-branch reclaim sweep (`isPathActive(task.worktree)`) — was +therefore blind to a live planner. Observed failure (FN-8600, 2026-07-26): the sweep saw a +zero-commit `fusion/fn-8600` whose tip trivially equals the integration ref, classified it +`tip-already-merged`, ran `git worktree remove --force` against the worktree a planning session was +running in, and escalated the resulting failure to `branch-conflict-unrecoverable` — parking a +healthy card `paused` with no operator action. A planner holding a worktree is a live session and +must be as visible as an executor or merger. +*/ +export type ActiveSessionKind = "executor" | "planning" | "step-session" | "workflow-step" | "step-session-parallel" | "ai-merge" | "workspace-repo-acquire" | "workspace-repo-land"; export interface ActiveSessionRegistration { taskId: string; diff --git a/packages/engine/src/concurrency.ts b/packages/engine/src/concurrency.ts index 62c054f4e4..5f36d4ef5c 100644 --- a/packages/engine/src/concurrency.ts +++ b/packages/engine/src/concurrency.ts @@ -125,32 +125,77 @@ export class ProjectAdmissionCoordinator { // Persisted task rows lag a fire-and-forget lane start, so omitting these // reservations lets a second coordinator pass over-admit one project. if (candidates.length === 0 || (await params.claimed()) + this.reservationCount(params.projectId) >= params.maxConcurrent) return; - const winner = candidates[0]; // Older test/runtime semaphore wrappers predate tryAcquire. They still // exercise project admission, while production semaphores atomically take // the host slot here. const hasReservableHostSlot = typeof params.semaphore?.tryAcquire === "function"; - const acquiredHostSlot = hasReservableHostSlot - ? params.semaphore!.tryAcquire() - : true; - if (!acquiredHostSlot) return; - // Compatibility-only semaphore shims cannot hold a reservation. Their - // lane tests provide claimed() synchronously, while real host semaphores - // use this durable marker until take/drop below. - if (hasReservableHostSlot) this.reserve(params.projectId, winner.taskId); - try { - winner.reserve?.(); - const accepted = await winner.start(); - if (accepted === false) { + /* + FNXC:ConcurrencyAdmission 2026-07-26-09:45: + Walk PAST a declining candidate instead of ending the pass on it. Previously only + `candidates[0]` was ever considered: when the oldest candidate's lane refused the handoff + (`start()` returning false — a merge id no longer in the queue, a duplicate/stale planner + claim), this returned having admitted nothing, and the next pass re-selected the same + still-declining winner. Oldest-first is a FAIRNESS order, not a permission for the oldest + candidate to veto every younger lane's work: a lane that cannot start is not consuming + capacity, so the slot belongs to the next candidate in age order. + + Ordering is preserved exactly — this only continues down the SAME sorted list, so a candidate + is never overtaken by a younger one that could have waited. Still at most ONE admission per + call, so no caller's capacity arithmetic changes. Each rejected attempt fully unwinds its own + reservation and host slot before the next is tried, which is why the acquire/reserve pair + moved inside the loop; a leak here would pin capacity permanently and is exactly the failure + this function exists to prevent. + */ + for (const winner of candidates) { + const acquiredHostSlot = hasReservableHostSlot + ? params.semaphore!.tryAcquire() + : true; + // The host semaphore is exhausted for everyone, not just this candidate: + // trying younger candidates cannot succeed, so stop rather than spin. + if (!acquiredHostSlot) return; + // Compatibility-only semaphore shims cannot hold a reservation. Their + // lane tests provide claimed() synchronously, while real host semaphores + // use this durable marker until take/drop below. + if (hasReservableHostSlot) this.reserve(params.projectId, winner.taskId); + /* + FNXC:ConcurrencyAdmission 2026-07-26-10:35: + Unwind EXACTLY what this attempt took. Two ways a naive `semaphore.release()` corrupts + accounting once declines are routine rather than pass-ending: + + 1. Compatibility shims without `tryAcquire` never acquire a slot and never get a reservation, + so releasing returns capacity nobody held — `returnSlot` decrements `_active` and drains a + waiter regardless. N decliners would free N phantom slots. + 2. `winner.reserve?.()` is `registerPreHeldExecutorSlot` for the triage and scheduler lanes. + Releasing the semaphore without dropping that registration leaves the id in the global + pre-held set with no backing acquire; the next pass's `takePreHeldExecutorSlot` then runs a + full top-level session without acquiring a slot and releases one it never held, leaving + `_active` permanently below the live agent count and the cap silently breached. + `dropPreHeldExecutorSlot` performs releaseReservation + release itself, but no-ops when + nothing was registered (the merge lane declines without reserving), so that case still needs + the explicit unwind. + */ + const releaseAttempt = () => { + if (!hasReservableHostSlot) return; + if (hasPreHeldExecutorSlot(winner.taskId)) { + dropPreHeldExecutorSlot(winner.taskId, params.semaphore); + return; + } this.releaseReservation(winner.taskId); params.semaphore?.release(); + }; + try { + winner.reserve?.(); + const accepted = await winner.start(); + if (accepted === false) { + releaseAttempt(); + continue; + } + admitted = winner.taskId; return; + } catch (error) { + releaseAttempt(); + throw error; } - admitted = winner.taskId; - } catch (error) { - this.releaseReservation(winner.taskId); - params.semaphore?.release(); - throw error; } })(); this.draining.set(params.projectId, drain); @@ -189,6 +234,25 @@ export const IDLE_SEMAPHORE_LEAK_REPAIR_MS = 5_000; * zombie in-progress rows kept the idle valve from ever firing) survives this * window trivially; a live nested reviewer does not. */ +/* +FNXC:GlobalConcurrencyControls 2026-07-26-11:15: +Deliberately still 600_000 after an attempt to lower it to 180_000 was reverted under review. +Motivation for the attempt: an operator-visible "Queued to plan" stall (FN-8600, 2026-07-26) lasted +~7 minutes, i.e. INSIDE this window, so the valve could not have fired even if leaked capacity was +the cause. Why it was reverted: nested runs -- the holders the original 600s was justified by -- are +already excluded from the reclaim floor (`reclaimFloor = bound + nestedActive`), so shortening the +window does not trade against them. What the window actually protects is a legitimate TOP-LEVEL +holder that `isRunningAgentTask` does not count: `runWithMergeAdmission` holds its slot for the whole +merge body (AI arbitration, post-merge audit, recovery dispatch, configured verification), including +the moments that park the row `paused`/`failed` or finalize it `done` -- all states that predicate +treats as not-running -- while neither lane's `inFlightCount` knows about a merge-lane slot. Inside a +shorter window the valve would reconcile that live slot away, `_drain()` would admit a replacement +over the cap, and the original body's later release would push `_active` permanently BELOW the true +live count (returnSlot clamps at 0 and never corrects upward) -- trading a bounded, visible stall for +an unbounded silent cap breach. Lower this only with measured evidence (the existing "recovered stale +semaphore active count X -> Y" warn gives the excess-duration distribution), or after merge-lane holds +are added to `inFlightCount` so a merge body can never look like leaked excess. +*/ export const STALE_SEMAPHORE_EXCESS_REPAIR_MS = 600_000; function createAbortError(): Error { diff --git a/packages/engine/src/run-audit.ts b/packages/engine/src/run-audit.ts index 32670aa3d9..f0c18136b9 100644 --- a/packages/engine/src/run-audit.ts +++ b/packages/engine/src/run-audit.ts @@ -466,6 +466,14 @@ export type DatabaseMutationType = | "task:checkout" /** Metadata: { taskId, artifactKeys, owner, source, action, attempt, maxAttempts, nodeId? } */ | "task:required-artifact-missing" + /** + * Planning admission was withheld because no top-level slot was reservable. + * Metadata: { blockedBy, maxConcurrent, claimed, projectRoom, eligibleCount, eligibleTaskIds, + * processingCount, processingTaskIds, semaphoreActiveCount?, semaphoreLimit?, + * semaphoreAvailableCount?, semaphoreWaitingCount? } — ids/counts/outcomes only. + * Deduped on the gate signature, so a sustained stall emits one row, not one per poll. + */ + | "task:plan-admission-throttled" | "agent:auto-recover-error-state" | "agent:reset-error-state-on-startup" | "agent:error-retry-exhausted" diff --git a/packages/engine/src/triage.ts b/packages/engine/src/triage.ts index a8be4b4ef9..8fcf4781b0 100644 --- a/packages/engine/src/triage.ts +++ b/packages/engine/src/triage.ts @@ -137,6 +137,7 @@ import { recoverIdleSemaphoreLeakCandidate, type AgentSemaphore, } from "./concurrency.js"; +import { acquireActiveSessionPath, activeSessionRegistry } from "./active-session-registry.js"; import { AgentLogger } from "./agent-logger.js"; import { resolveAgentInstructions, @@ -271,6 +272,13 @@ export class TriageProcessor { private wasGlobalPaused = false; private wasEnginePaused = false; private idleSemaphoreLeakCandidateSince: number | null = null; + /** + * FNXC:ConcurrencyAdmission 2026-07-26-09:30: + * Signature of the last emitted `task:plan-admission-throttled` event, so a steady stall records + * one row instead of one per poll. `null` means "not currently throttled" — the next throttle, + * even with identical numbers, is a NEW stall and is emitted again. + */ + private lastPlanThrottleSignature: string | null = null; /** Active agent sessions per task, used to terminate on pause. */ private activeSessions = new Map void }>(); /** @@ -1489,6 +1497,90 @@ export class TriageProcessor { `maxConcurrent=${maxConcurrent}, claimed=${claimed}, processing=${this.processing.size}` + `${processingIds.length > 0 ? ` [${processingIds.join(", ")}]` : ""}${semaphoreDetail}`, ); + /* + FNXC:ConcurrencyAdmission 2026-07-26-09:30: + Durable counterpart to the log line above. Requirement from a real incident (FN-8600, + 2026-07-26): an operator asked why a started card sat "Queued to plan" for seven minutes, and + it was UNANSWERABLE after the fact — the binding gate existed only in this `planLog.log`, + which lands in the TUI's in-memory pane (truncated to ~40 chars) and is persisted nowhere. + Reconstructing it cost a full DB forensics pass and still could not separate "host semaphore + exhausted" from "project cap consumed". Emitting the gate to run-audit makes it answerable at + all. Caveat: today the only run-audit READ route resolves through a durable agent's heartbeat + run, and this event carries a synthetic run id under agentId "triage", so it is reachable by + direct DB query but not yet through any dashboard route or fn_* tool -- the same blind spot + every synthetic-run self-healing/scheduler diagnostic shares. A task/type-scoped run-audit + read surface would close it for all of them at once. + + Metadata stays ids/counts/outcomes-only per the run-audit contract: gate name, caps, counts, + and at most five eligible/processing task IDs — never prompts, titles, or reasons prose. + + Deduped on the gate signature, not on time: while the gate and the cards behind it hold + steady a long stall collapses to ONE row instead of ~28 at a 15s poll, which is what keeps + operators reading the event. The signature deliberately includes the eligible task IDs -- + counts alone would let a NEW card's stall be swallowed whenever the numbers happened to land + on the same tuple, and "why is THIS card queued" is the question the event exists to answer. + Live counts still jitter as unrelated lanes cycle, so this bounds write volume rather than + guaranteeing exactly one row. + */ + const throttleSignature = [ + blockedBy, + maxConcurrent, + claimed, + triageTasks.length, + this.processing.size, + semaphoreSnapshot?.activeCount ?? -1, + semaphoreSnapshot?.limit ?? -1, + eligibleIds.join(","), + ].join("|"); + if (this.lastPlanThrottleSignature !== throttleSignature) { + const throttleAuditor = createRunAuditor(this.store, { + taskId: eligibleIds[0], + agentId: "triage", + runId: generateSyntheticRunId("plan-admission-throttled", eligibleIds[0] ?? this.rootDir), + phase: "triage", + source: "triage", + }); + /* + FNXC:ConcurrencyAdmission 2026-07-26-10:45: + Fire-and-forget. Awaiting a store write inside the 15s poll let a slow/hung write delay the + NEXT poll's chance to notice freed capacity -- compounding the very stall being recorded. + */ + void throttleAuditor.database({ + type: "task:plan-admission-throttled", + target: eligibleIds[0] ?? this.rootDir, + metadata: { + blockedBy, + maxConcurrent, + claimed, + projectRoom, + eligibleCount: triageTasks.length, + eligibleTaskIds: eligibleIds, + processingCount: this.processing.size, + processingTaskIds: processingIds, + semaphoreActiveCount: semaphoreSnapshot?.activeCount, + semaphoreLimit: semaphoreSnapshot?.limit, + semaphoreAvailableCount: semaphoreSnapshot?.availableCount, + semaphoreWaitingCount: semaphoreSnapshot?.waitingCount, + }, + }) + .then(() => { + /* + FNXC:ConcurrencyAdmission 2026-07-26-10:45: + Mark the stall as recorded ONLY once the write lands. Setting the marker up front meant + a failed write -- most likely exactly when the store is contended, the condition this + event is meant to explain -- was swallowed for the whole stall with no retry, leaving + the incident as unanswerable as before the event existed. On failure the marker stays + put so the next poll retries. + */ + this.lastPlanThrottleSignature = throttleSignature; + }) + .catch((auditErr: unknown) => { + planLog.warn(`Failed to write plan-admission-throttled run-audit event: ${auditErr instanceof Error ? auditErr.message : String(auditErr)}`); + }); + } + } else { + // Capacity is available again — the next distinct stall must re-announce itself. + this.lastPlanThrottleSignature = null; } // Keep handoff reservations visible even when a test/runtime wrapper delays @@ -1597,6 +1689,14 @@ export class TriageProcessor { this.processing.add(task.id); this.processingSince.set(task.id, Date.now()); + /* + FNXC:NodeWorktreeIsolation 2026-07-26-09:10: + Holds the worktree path this planning run published to `activeSessionRegistry`, so the outer + `finally` can release exactly what it registered (and nothing when planning ran in the shared + checkout). Declared at method scope because registration happens deep inside the try. + */ + let registeredPlanningPath: string | null = null; + planLog.log( `Specifying ${task.id}: ${task.title || task.description.slice(0, 60)}`, ); @@ -1907,8 +2007,50 @@ export class TriageProcessor { same shared-path shape behind the reported Plan Review session collision. The worktree acquired here is the one Plan Review and the implementation session then reuse. */ - const planningCwd = (await this.options.acquirePlanningWorktree?.(task.id).catch(() => null)) || this.rootDir; + let planningCwd = (await this.options.acquirePlanningWorktree?.(task.id).catch(() => null)) || this.rootDir; if (planningCwd !== this.rootDir) { + /* + FNXC:NodeWorktreeIsolation 2026-07-26-09:10: + Publish the planner's worktree as a live session for as long as this planning run owns it. + Registration is what makes `activeSessionRegistry.isPathActive()` true, which is the single + guard every worktree-removal path consults. Without it the self-healing reclaim sweep tore + the worktree out from under a running planner and paused the task + `branch-conflict-unrecoverable` (FN-8600). Registered ONLY for a real task worktree — the + shared `rootDir` fallback is the operator's checkout and must never be marked task-owned. + The reciprocal unregister lives in this method's outer `finally`, so an early throw between + here and there cannot leak a permanent entry that blocks later legitimate cleanup. + */ + /* + FNXC:NodeWorktreeIsolation 2026-07-26-10:25: + Acquire through the reclaim-aware seam, not raw `registerPath`. `acquireActiveSessionPath` + is what lets a leaked entry from a crashed/dead holder be reclaimed instead of hard-failing + a legitimate new registration; raw `registerPath` throws on any foreign-held path, so one + stale record would wedge planning for that worktree until the engine restarted. The + executor already registers exclusively through this seam + (`TaskExecutor.acquireSessionRegistryPath`) — planning must not be the one holder that + bypasses it. `contended` means a genuinely live foreign holder, so planning falls back to + the shared checkout rather than running in a worktree someone else owns. + */ + const acquired = acquireActiveSessionPath(activeSessionRegistry, planningCwd, { + taskId: task.id, + kind: "planning", + ownerKey: `planning:${task.id}`, + }, { + holderLiveProbe: (holderTaskId) => this.processing.has(holderTaskId) || this.hasLivePlanningWork(holderTaskId), + }); + if (acquired.action === "contended") { + planLog.warn( + `${task.id}: planning worktree ${planningCwd} is held by live task ${acquired.holderTaskId} (${acquired.holderKind}) — planning in the shared checkout instead`, + ); + planningCwd = this.rootDir; + } else { + if (acquired.action === "reclaimed-stale-foreign") { + planLog.warn( + `${task.id}: reclaimed a stale active-session entry on ${planningCwd} from dead task ${acquired.holderTaskId} (idle ${acquired.ageMs}ms)`, + ); + } + registeredPlanningPath = planningCwd; + } await this.store.logEntry(task.id, `Planning session running in task worktree ${planningCwd}`).catch(() => undefined); } const { session } = await createResolvedAgentSession({ @@ -2446,6 +2588,32 @@ export class TriageProcessor { // early setup failure must return that untransferred host slot; after a // successful transfer this is intentionally a no-op. dropPreHeldExecutorSlot(task.id, this.options.semaphore); + /* + FNXC:NodeWorktreeIsolation 2026-07-26-09:10: + Release the planner's registry entry on EVERY exit path (success, planning failure, abort, + pause). A leaked entry is not merely untidy: `isPathActive` would stay true forever and + permanently veto legitimate reclaim/cleanup of that worktree, converting this fix into the + opposite stall. Unregister is keyed on what we registered, so it is a no-op for planning runs + that used the shared checkout. + */ + if (registeredPlanningPath) { + /* + FNXC:NodeWorktreeIsolation 2026-07-26-10:20: + Release ONLY if this planning run still owns the record. A bare path delete would reintroduce + this fix's own symptom on the execution side: `finalizeApprovedTask` moves the card to `todo` + while still inside the try, and several awaited writes (log flush, token-usage record, + getTask/updateTask, dispose) run before this finally. The scheduler can dispatch in that + window and the executor registers the SAME worktree path — `registerPath` permits a same-task + overwrite, so the record becomes `kind:"executor"`. Deleting by path alone would then clear a + LIVE executor entry, making `isPathActive` false and handing the reclaim sweep the same + worktree it tore out from under a planner in FN-8600. + */ + const record = activeSessionRegistry.lookupByPath(registeredPlanningPath); + if (record?.ownerKey === `planning:${task.id}`) { + activeSessionRegistry.unregisterPath(registeredPlanningPath); + } + registeredPlanningPath = null; + } this.processing.delete(task.id); this.processingSince.delete(task.id); this.coordinatorAdmittedTaskIds.delete(task.id);