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

fix(FN-8600): keep self-healing from pausing live planners and unstick queued planning

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
gsxdsm
2026-07-26 09:26:13 -07:00
parent d4aa79b66c
commit 2d263acc49
9 changed files with 974 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
.changeset/planning-worktree-reclaim-race.md Normal file
View File

@@ -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.

1
AGENTS.md
View File

@@ -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: <reason>"`) 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.

172
packages/engine/src/__tests__/concurrency.test.ts
View File

@@ -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<ProjectAdmissionCoordinator["admitOldest"]>[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);

236
packages/engine/src/__tests__/triage-plan-admission-throttle-audit.test.ts Normal file
View File

@@ -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<typeof import("@fusion/core")>();
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<string, unknown> }
function createStore(tasks: Task[], recorded: RecordedEvent[], settings: Partial<Settings> = {}): 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<string, unknown> }) => {
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<TriageProcessor> {
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<void> }).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<void> }).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<void> }).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<string, unknown> }) => {
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<void> }).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();
});
});

285
packages/engine/src/__tests__/triage-planning-worktree-session-registration.test.ts Normal file
View File

@@ -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<typeof import("@fusion/core")>();
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> = {}): 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);
});
});

15
packages/engine/src/active-session-registry.ts
View File

@@ -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;

100
packages/engine/src/concurrency.ts
View File

@@ -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 {

8
packages/engine/src/run-audit.ts
View File

@@ -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"

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

@@ -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<string, { dispose: () => 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);