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

feat(FN-4954): harden pool leasing invariants and wire runtime auditing

Browse Source 
Fusion-Task-Id: FN-4954
Fusion-Task-Lineage: b3dd2d2f-3aa3-48f1-8a56-56e66518d139
This commit is contained in:
Fusion (runfusion.ai)
2026-05-17 20:22:06 -07:00
committed by gsxdsm
parent 0606e3aaea
commit 7e90bce8bb
8 changed files with 220 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
packages/engine/src/__tests__/restart.integration.test.ts
View File

@@ -1577,7 +1577,7 @@ describe("Edge case: worktree deleted between scan and acquire", () => {
mockedExistsSync.mockReturnValue(false);
// acquire() checks existsSync and prunes the stale entry
const result = pool.acquire();
const result = pool.acquire("FN-test");
expect(result).toBeNull();
expect(pool.size).toBe(0);
});

8
packages/engine/src/__tests__/worktree-acquisition.test.ts
View File

@@ -54,7 +54,7 @@ describe("acquireTaskWorktree", () => {
store,
settings: { recycleWorktrees: true } as any,
pool: {
acquire: () => "/tmp/pooled",
acquire: (_taskId: string) => "/tmp/pooled",
prepareForTask,
release,
} as any,
@@ -79,7 +79,7 @@ describe("acquireTaskWorktree", () => {
store,
settings: { recycleWorktrees: true } as any,
pool: {
acquire: () => "/tmp/pooled",
acquire: (_taskId: string) => "/tmp/pooled",
prepareForTask: vi.fn().mockResolvedValue({
branch: "fusion/fn-1",
worktreePath: "/tmp/live-existing",
@@ -92,7 +92,7 @@ describe("acquireTaskWorktree", () => {
createWorktree: vi.fn(),
});
expect(release).toHaveBeenCalledWith("/tmp/pooled");
expect(release).toHaveBeenCalledWith("/tmp/pooled", "FN-1");
});
it("falls through to fresh creation when pooled worktree is incomplete and emits detection audit", async () => {
@@ -107,7 +107,7 @@ describe("acquireTaskWorktree", () => {
store,
settings: { recycleWorktrees: true } as any,
pool: {
acquire: () => "/tmp/pooled",
acquire: (_taskId: string) => "/tmp/pooled",
prepareForTask: vi.fn().mockResolvedValue({ branch: "fusion/fn-1", worktreePath: "/tmp/pooled", reclaimed: false }),
release: vi.fn(),
} as any,

35
packages/engine/src/__tests__/worktree-pool-double-lease.test.ts
View File

@@ -10,25 +10,42 @@ vi.mock("node:fs", () => ({
import { WorktreePool } from "../worktree-pool.js";
// FN-4954: deterministic repro for the rehydrate collision race.
describe("WorktreePool double-lease reproduction", () => {
// FN-4954 deterministic race backstop.
describe("WorktreePool double-lease guard", () => {
let pool: WorktreePool;
beforeEach(() => {
pool = new WorktreePool();
});
it("reproduces rehydrate re-adding a path that is already leased", () => {
pool.release("/tmp/wt-race");
it("prevents rehydrate from re-adding a leased path", () => {
const violations: Array<{ phase: string; existingHolder: string }> = [];
pool.setInvariantViolationHandler((violation) => {
violations.push({ phase: violation.phase, existingHolder: violation.existingHolder });
});
const firstLease = pool.acquire();
pool.release("/tmp/wt-race");
const firstLease = pool.acquire("FN-A");
expect(firstLease).toBe("/tmp/wt-race");
// Simulates scan/rehydrate colliding with an in-flight lease.
pool.rehydrate(["/tmp/wt-race"]);
// Expected invariant: leased paths must never be re-added to idle.
// Current behavior (pre-fix) returns the same path again here.
expect(pool.acquire()).toBeNull();
expect(pool.acquire("FN-B")).toBeNull();
expect(pool.size).toBe(0);
expect(pool.getLeasedPaths().get("/tmp/wt-race")).toBe("FN-A");
expect(violations).toEqual([{ phase: "rehydrate", existingHolder: "FN-A" }]);
});
it("keeps release best-effort when releasing task differs", () => {
const violations: Array<{ phase: string; requestingTaskId: string }> = [];
pool.setInvariantViolationHandler((violation) => violations.push({ phase: violation.phase, requestingTaskId: violation.requestingTaskId }));
pool.release("/tmp/wt-race");
expect(pool.acquire("FN-A")).toBe("/tmp/wt-race");
pool.release("/tmp/wt-race", "FN-B");
expect(pool.has("/tmp/wt-race")).toBe(true);
expect(pool.getLeasedPaths().size).toBe(0);
expect(violations).toEqual([{ phase: "release", requestingTaskId: "FN-B" }]);
});
});

34
packages/engine/src/__tests__/worktree-pool.test.ts
View File

@@ -66,6 +66,7 @@ const mockedExistsSync = vi.mocked(existsSync);
const mockedLstatSync = vi.mocked(lstatSync);
const mockedReaddirSync = vi.mocked(readdirSync);
const mockedRmSync = vi.mocked(rmSync);
const TEST_TASK_ID = "FN-test";
let errorSpy: ReturnType<typeof vi.spyOn>;
let warnSpy: ReturnType<typeof vi.spyOn>;
@@ -91,12 +92,12 @@ describe("WorktreePool", () => {
describe("acquire", () => {
it("returns null when pool is empty", () => {
expect(pool.acquire()).toBeNull();
expect(pool.acquire(TEST_TASK_ID)).toBeNull();
});
it("returns a released path on acquire", () => {
pool.release("/tmp/worktree-1");
const result = pool.acquire();
const result = pool.acquire(TEST_TASK_ID);
expect(result).toBe("/tmp/worktree-1");
});
@@ -106,7 +107,7 @@ describe("WorktreePool", () => {
// First path doesn't exist, second does
mockedExistsSync.mockImplementation((p) => p === "/tmp/good-worktree");
const result = pool.acquire();
const result = pool.acquire(TEST_TASK_ID);
expect(result).toBe("/tmp/good-worktree");
expect(pool.size).toBe(0);
});
@@ -116,11 +117,30 @@ describe("WorktreePool", () => {
pool.release("/tmp/stale-2");
mockedExistsSync.mockReturnValue(false);
expect(pool.acquire()).toBeNull();
expect(pool.acquire(TEST_TASK_ID)).toBeNull();
expect(pool.size).toBe(0);
});
});
describe("double-lease invariant", () => {
it("skips rehydrate entries that are already leased", () => {
const handler = vi.fn();
pool.setInvariantViolationHandler(handler);
pool.release("/tmp/wt-lease");
expect(pool.acquire(TEST_TASK_ID)).toBe("/tmp/wt-lease");
pool.rehydrate(["/tmp/wt-lease"]);
expect(pool.size).toBe(0);
expect(pool.getLeasedPaths().get("/tmp/wt-lease")).toBe(TEST_TASK_ID);
expect(handler).toHaveBeenCalledWith(expect.objectContaining({
path: "/tmp/wt-lease",
existingHolder: TEST_TASK_ID,
phase: "rehydrate",
}));
});
});
describe("release", () => {
it("adds a path to the pool", () => {
pool.release("/tmp/wt-1");
@@ -141,9 +161,9 @@ describe("WorktreePool", () => {
pool.release("/tmp/a");
pool.release("/tmp/b");
expect(pool.size).toBe(2);
pool.acquire();
pool.acquire(TEST_TASK_ID);
expect(pool.size).toBe(1);
pool.acquire();
pool.acquire(TEST_TASK_ID);
expect(pool.size).toBe(0);
});
});
@@ -160,7 +180,7 @@ describe("WorktreePool", () => {
it("returns false after path is acquired", () => {
pool.release("/tmp/wt");
pool.acquire();
pool.acquire(TEST_TASK_ID);
expect(pool.has("/tmp/wt")).toBe(false);
});
});

37
packages/engine/src/merger.ts
View File

@@ -7776,15 +7776,34 @@ export async function aiMergeTask(
mergerLog.log(`Worktree retained — still needed by ${otherUser}`);
result.worktreeRemoved = false;
} else if (options.pool && settings.recycleWorktrees) {
options.pool.release(worktreePath);
result.worktreeRemoved = false;
// Detach the path from this task so future diff queries don't read
// a foreign branch's state once the pool reassigns this worktree.
try {
await store.updateTask(taskId, { worktree: null, branch: null });
} catch (err: unknown) {
const msg = err instanceof Error ? err.message : String(err);
mergerLog.warn(`${taskId}: failed to clear worktree pointer after pool release: ${msg}`);
if (activeSessionRegistry.isPathActive(worktreePath)) {
mergerLog.warn(`${taskId}: skipping pooled release for active session path ${worktreePath}`);
await audit?.git({
type: "worktree:removal-refused-active-session",
target: worktreePath,
metadata: { taskId, reason: RemovalReason.MergerCleanup, kind: "merger" },
});
result.worktreeRemoved = false;
} else {
try {
const onBranch = await execAsync("git symbolic-ref --quiet HEAD", { cwd: worktreePath, timeout: 5_000, encoding: "utf-8" })
.then(() => true)
.catch(() => false);
if (onBranch) {
await execAsync("git checkout --detach HEAD", { cwd: worktreePath, timeout: 10_000, encoding: "utf-8" });
}
} catch (err: unknown) {
const msg = err instanceof Error ? err.message : String(err);
mergerLog.warn(`${taskId}: failed to detach pooled worktree before release: ${msg}`);
}
try {
await store.updateTask(taskId, { worktree: null, branch: null });
} catch (err: unknown) {
const msg = err instanceof Error ? err.message : String(err);
mergerLog.warn(`${taskId}: failed to clear worktree pointer before pool release: ${msg}`);
}
options.pool.release(worktreePath, taskId);
result.worktreeRemoved = false;
}
} else {
try {

30
packages/engine/src/runtimes/in-process-runtime.ts
View File

@@ -16,7 +16,7 @@ import { Scheduler } from "../scheduler.js";
import type { PrMonitor, PrComment } from "../pr-monitor.js";
import type { PrInfo } from "@fusion/core";
import { TaskExecutor, type TaskExecutorOptions } from "../executor.js";
import { WorktreePool, isGitRepository } from "../worktree-pool.js";
import { WorktreePool, isGitRepository, type PoolInvariantViolation } from "../worktree-pool.js";
import { AgentSemaphore } from "../concurrency.js";
import { HeartbeatMonitor, HeartbeatTriggerScheduler, type WakeContext } from "../agent-heartbeat.js";
import { AutoClaimSnapshotManager } from "../auto-claim-snapshot.js";
@@ -43,6 +43,7 @@ import { TriageProcessor } from "../triage.js";
import { EphemeralWorkerManager } from "../ephemeral-worker-manager.js";
import { validateProjectNodeMapping } from "../node-dispatch-validation.js";
import { attachAgentLinkSync } from "../task-agent-sync.js";
import { createRunAuditor, generateSyntheticRunId } from "../run-audit.js";
/**
* InProcessRuntime runs a project within the main process.
@@ -469,6 +470,33 @@ export class InProcessRuntime
executorOptions
);
this.worktreePool.setInvariantViolationHandler((violation: PoolInvariantViolation) => {
void (async () => {
try {
runtimeLog.warn(
`[worktree-pool] invariant violation detected (${violation.phase}) path=${violation.path} holder=${violation.existingHolder} requester=${violation.requestingTaskId}`,
);
const audit = createRunAuditor(this.taskStore, {
runId: generateSyntheticRunId("worktree-pool-invariant", violation.requestingTaskId),
taskId: violation.requestingTaskId,
agentId: "system",
phase: "execute",
});
await audit.db({
type: "worktree:pool-double-lease-detected",
target: violation.path,
metadata: violation,
});
await this.taskStore.logEntry(
violation.requestingTaskId,
`Worktree pool invariant violation (${violation.phase}): ${violation.path} is held by ${violation.existingHolder}`,
);
} catch (error) {
runtimeLog.warn(`Failed to process worktree pool invariant violation: ${error instanceof Error ? error.message : String(error)}`);
}
})();
});
// 6. Initialize HeartbeatMonitor (reuses AgentStore from step 5a)
if (this.heartbeatMonitor) {
// Already started — nothing to do

24
packages/engine/src/worktree-acquisition.ts
View File

@@ -13,6 +13,7 @@ import {
isInsideWorktreesDir,
removeWorktree,
RemovalReason,
PoolDoubleLeaseError,
} from "./worktree-pool.js";
import {
NativeWorktreeBackend,
@@ -257,7 +258,18 @@ export async function acquireTaskWorktree(opts: AcquireTaskWorktreeOptions): Pro
let branch = branchName;
if (!isResume && pool && settings.recycleWorktrees) {
const pooled = pool.acquire();
let pooled: string | null = null;
try {
pooled = pool.acquire(task.id);
} catch (poolErr) {
if (poolErr instanceof PoolDoubleLeaseError) {
const poolErrMessage = poolErr instanceof Error ? poolErr.message : String(poolErr);
logger?.warn(`${task.id}: ${poolErrMessage}; skipping pool and creating fresh worktree`);
await store.logEntry(task.id, `Pool double-lease guard triggered (${poolErrMessage}), creating fresh worktree`, undefined, runContext);
} else {
throw poolErr;
}
}
if (pooled) {
try {
const preparedRaw = await pool.prepareForTask(pooled, branchName, baseBranch ?? undefined, {
@@ -269,7 +281,7 @@ export async function acquireTaskWorktree(opts: AcquireTaskWorktreeOptions): Pro
? { branch: preparedRaw, worktreePath: pooled, reclaimed: false as const }
: preparedRaw;
if (prepared.reclaimed && prepared.worktreePath !== pooled) {
pool.release(pooled);
pool.release(pooled, task.id);
}
worktreePath = prepared.worktreePath;
branch = prepared.branch;
@@ -341,8 +353,12 @@ export async function acquireTaskWorktree(opts: AcquireTaskWorktreeOptions): Pro
};
}
} catch (poolErr) {
pool.release(pooled);
if (isBranchConflictError(poolErr)) throw poolErr;
pool.release(pooled, task.id);
if (poolErr instanceof PoolDoubleLeaseError) {
const poolErrMessage = poolErr instanceof Error ? poolErr.message : String(poolErr);
logger?.warn(`${task.id}: ${poolErrMessage}; skipping pool and creating fresh worktree`);
await store.logEntry(task.id, `Pool double-lease guard triggered (${poolErrMessage}), creating fresh worktree`, undefined, runContext);
} else if (isBranchConflictError(poolErr)) throw poolErr;
const poolErrMessage = poolErr instanceof Error ? poolErr.message : String(poolErr);
logger?.log(`Pool prepareForTask failed, falling through to fresh worktree: ${poolErrMessage}`);
await store.logEntry(task.id, `Pool worktree preparation failed (${poolErrMessage}), creating fresh worktree`, undefined, runContext);

90
packages/engine/src/worktree-pool.ts
View File

@@ -241,8 +241,31 @@ export type PrepareForTaskResult = {
strandedCommitCount?: number;
};
export type PoolInvariantPhase = "acquire" | "rehydrate" | "release";
export type PoolInvariantViolation = {
path: string;
existingHolder: string;
requestingTaskId: string;
phase: PoolInvariantPhase;
};
export class PoolDoubleLeaseError extends Error {
constructor(
public readonly path: string,
public readonly existingHolder: string,
public readonly requestingTaskId: string,
public readonly phase: PoolInvariantPhase,
) {
super(`Pool double lease detected for ${path}: held by ${existingHolder}, requested by ${requestingTaskId} during ${phase}`);
this.name = "PoolDoubleLeaseError";
}
}
export class WorktreePool {
private idle = new Set<string>();
private leased = new Map<string, string>();
private invariantViolationHandler?: (violation: PoolInvariantViolation) => void;
/**
* Acquire an idle worktree from the pool.
@@ -251,12 +274,15 @@ export class WorktreePool {
* is empty. Before returning, verifies the directory still exists on disk
* and prunes any stale entries.
*/
acquire(): string | null {
acquire(taskId: string): string | null {
for (const path of this.idle) {
this.assertNotDoubleLeased(path, taskId, "acquire");
this.idle.delete(path);
this.leased.set(path, taskId);
if (existsSync(path)) {
return path;
}
this.leased.delete(path);
worktreePoolLog.log(`Pruned stale entry: ${path}`);
}
return null;
@@ -270,7 +296,22 @@ export class WorktreePool {
*
* @param worktreePath — Absolute path to the worktree directory
*/
release(worktreePath: string): void {
release(worktreePath: string, releasingTaskId?: string): void {
const existingHolder = this.leased.get(worktreePath);
if (!existingHolder) {
worktreePoolLog.warn(`release called for non-leased worktree: ${worktreePath}`);
} else if (releasingTaskId && existingHolder !== releasingTaskId) {
this.notifyInvariantViolation({
path: worktreePath,
existingHolder,
requestingTaskId: releasingTaskId,
phase: "release",
});
worktreePoolLog.warn(
`release task mismatch for ${worktreePath}: leased holder=${existingHolder}, releasingTaskId=${releasingTaskId}`,
);
}
this.leased.delete(worktreePath);
this.idle.add(worktreePath);
}
@@ -284,6 +325,33 @@ export class WorktreePool {
return this.idle.has(path);
}
setInvariantViolationHandler(handler: (violation: PoolInvariantViolation) => void): void {
this.invariantViolationHandler = handler;
}
/** @internal test-only visibility */
getLeasedPaths(): ReadonlyMap<string, string> {
return this.leased;
}
private notifyInvariantViolation(violation: PoolInvariantViolation): void {
try {
this.invariantViolationHandler?.(violation);
} catch (error) {
worktreePoolLog.warn(`Invariant violation handler failed: ${error instanceof Error ? error.message : String(error)}`);
}
}
private assertNotDoubleLeased(path: string, requestingTaskId: string, phase: PoolInvariantPhase): void {
const existingHolder = this.leased.get(path);
if (!existingHolder || existingHolder === requestingTaskId) {
return;
}
const violation: PoolInvariantViolation = { path, existingHolder, requestingTaskId, phase };
this.notifyInvariantViolation(violation);
throw new PoolDoubleLeaseError(path, existingHolder, requestingTaskId, phase);
}
/**
* Remove and return all idle worktree paths.
*
@@ -293,6 +361,7 @@ export class WorktreePool {
drain(): string[] {
const paths = Array.from(this.idle);
this.idle.clear();
this.leased.clear();
return paths;
}
@@ -306,11 +375,22 @@ export class WorktreePool {
*/
rehydrate(idlePaths: string[]): void {
for (const path of idlePaths) {
if (existsSync(path)) {
this.idle.add(path);
} else {
if (!existsSync(path)) {
worktreePoolLog.log(`Rehydrate skipped (not on disk): ${path}`);
continue;
}
const existingHolder = this.leased.get(path);
if (existingHolder) {
this.notifyInvariantViolation({
path,
existingHolder,
requestingTaskId: existingHolder,
phase: "rehydrate",
});
worktreePoolLog.warn(`Rehydrate skipped leased worktree ${path} (holder=${existingHolder})`);
continue;
}
this.idle.add(path);
}
}