diff --git a/.changeset/fn-8105-archive-removes-worktree.md b/.changeset/fn-8105-archive-removes-worktree.md new file mode 100644 index 0000000000..f08b03712e --- /dev/null +++ b/.changeset/fn-8105-archive-removes-worktree.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Archiving a task now deletes its git worktree so pinned worktrees no longer leak. +category: fix +dev: Archive cleanup uses a store-scoped engine disposer and host-scoped worktree-path reservation; cleanup:false retains the worktree and workspace per-repo cleanup remains deferred. diff --git a/docs/task-management.md b/docs/task-management.md index 817a232876..a96dc47b18 100644 --- a/docs/task-management.md +++ b/docs/task-management.md @@ -355,6 +355,10 @@ Auto-completion/finalization remains owned by existing recovery passes: 5. **done** — merged/finalized 6. **archived** — preserved history, optionally cleaned from filesystem +### Archive worktree cleanup + +Archiving a single-repository task synchronously removes its git worktree before branch and task-metadata cleanup. This applies to random and pinned (`task-id`/`task-title`) names, including `fn_task_archive` and direct CLI archive commands that run without an executor. A host-scoped filesystem reservation serializes a successor's deterministic-path acquisition with archival disposal; if removal fails, the reservation is quarantined so the next acquisition can reconcile the orphan instead of colliding with it. `archive({ cleanup: false })` intentionally retains the worktree. Workspace tasks' per-repository `workspaceWorktrees` are not removed by this lifecycle yet. + Board ordering behavior: - `todo` mirrors scheduler dispatch order: priority first (`urgent` → `low`), then oldest `createdAt` within a priority tier, then task ID as deterministic tie-break. - `triage`, `in-progress`, and `in-review` remain priority-first with task-ID tie-breaks (`in-review` still pins merge-active statuses above non-merging tasks). diff --git a/packages/cli/src/__tests__/extension-experiment-finalize.test.ts b/packages/cli/src/__tests__/extension-experiment-finalize.test.ts index cb8f2d6ddd..60337e5196 100644 --- a/packages/cli/src/__tests__/extension-experiment-finalize.test.ts +++ b/packages/cli/src/__tests__/extension-experiment-finalize.test.ts @@ -62,6 +62,7 @@ vi.mock("@fusion/dashboard", () => ({ })); vi.mock("@fusion/engine", () => ({ + installBaselineArchiveWorktreeDisposer: vi.fn(), ...workflowAuthoringEngineMock, createFnAgent: vi.fn(), fetchWebContent: vi.fn(), diff --git a/packages/cli/src/__tests__/extension-fn-secret-get.test.ts b/packages/cli/src/__tests__/extension-fn-secret-get.test.ts index ab55b80642..2836705f38 100644 --- a/packages/cli/src/__tests__/extension-fn-secret-get.test.ts +++ b/packages/cli/src/__tests__/extension-fn-secret-get.test.ts @@ -19,6 +19,7 @@ vi.mock("@fusion/dashboard", () => ({ buildGitLabTaskDescription: vi.fn(), })); vi.mock("@fusion/engine", () => ({ + installBaselineArchiveWorktreeDisposer: vi.fn(), ...workflowAuthoringEngineMock, createFnAgent: vi.fn(), fetchWebContent: vi.fn(), diff --git a/packages/cli/src/__tests__/extension-gitlab-tracking.test.ts b/packages/cli/src/__tests__/extension-gitlab-tracking.test.ts index cb028df620..597a6aa447 100644 --- a/packages/cli/src/__tests__/extension-gitlab-tracking.test.ts +++ b/packages/cli/src/__tests__/extension-gitlab-tracking.test.ts @@ -36,6 +36,7 @@ vi.mock("@fusion/dashboard", () => { }); vi.mock("@fusion/engine", () => ({ + installBaselineArchiveWorktreeDisposer: vi.fn(), createFnAgent: vi.fn(), fetchWebContent: vi.fn(), assertNoSecretPlaintext: vi.fn(), diff --git a/packages/cli/src/__tests__/extension-web-fetch.test.ts b/packages/cli/src/__tests__/extension-web-fetch.test.ts index 844faeae81..6b0d782cd2 100644 --- a/packages/cli/src/__tests__/extension-web-fetch.test.ts +++ b/packages/cli/src/__tests__/extension-web-fetch.test.ts @@ -14,6 +14,7 @@ vi.mock("@fusion/dashboard", () => ({ })); vi.mock("@fusion/engine", () => ({ + installBaselineArchiveWorktreeDisposer: vi.fn(), ...workflowAuthoringEngineMock, createFnAgent: vi.fn(), fetchWebContent: fetchWebContentMock, diff --git a/packages/cli/src/__tests__/task-command-github-import-tracking.test.ts b/packages/cli/src/__tests__/task-command-github-import-tracking.test.ts index dd0e323a11..1344fe2c31 100644 --- a/packages/cli/src/__tests__/task-command-github-import-tracking.test.ts +++ b/packages/cli/src/__tests__/task-command-github-import-tracking.test.ts @@ -57,6 +57,7 @@ vi.mock("@fusion/dashboard", () => ({ })); vi.mock("@fusion/engine", () => ({ + installBaselineArchiveWorktreeDisposer: vi.fn(), createFnAgent: vi.fn(), runAiMerge: vi.fn(), landWorkspaceTask: vi.fn(), diff --git a/packages/cli/src/commands/__tests__/task.test.ts b/packages/cli/src/commands/__tests__/task.test.ts index f6f3f0a80b..7f242b69ee 100644 --- a/packages/cli/src/commands/__tests__/task.test.ts +++ b/packages/cli/src/commands/__tests__/task.test.ts @@ -97,6 +97,7 @@ vi.mock("@fusion/core", async (importActual) => { // Mock @fusion/engine vi.mock("@fusion/engine", () => ({ + installBaselineArchiveWorktreeDisposer: vi.fn(), aiMergeTask: vi.fn(), runAiMerge: vi.fn(), landWorkspaceTask: vi.fn(), diff --git a/packages/cli/src/commands/task.ts b/packages/cli/src/commands/task.ts index 2c55463f28..95fde60359 100644 --- a/packages/cli/src/commands/task.ts +++ b/packages/cli/src/commands/task.ts @@ -1,5 +1,5 @@ import { TaskStore, COLUMNS, COLUMN_LABELS, CentralCore, buildAutoPauseClearPatch, buildManualRetryResetPatch, extractIntentSignature, findNearDuplicates, getTaskDuplicateLineage, isWorkspaceTask, reconcileDeterministicDuplicate, resolveTaskGithubTracking, runDeterministicDuplicateGuard, type Settings, type Column, type ColumnId, type StepStatus, type AgentLogType, type AgentLogEntry, type IntentSignature, type NearDuplicateCandidate, type NearDuplicateMatch, type TaskDependencyMutation } from "@fusion/core"; -import { isInReviewMissingWorktreeSessionStartFailure, runAiMerge, landWorkspaceTask } from "@fusion/engine"; +import { isInReviewMissingWorktreeSessionStartFailure, runAiMerge, landWorkspaceTask, installBaselineArchiveWorktreeDisposer } from "@fusion/engine"; import { createInterface } from "node:readline/promises"; import type { PlanningQuestion, PlanningSummary } from "@fusion/core"; import { createSession, submitResponse, RateLimitError, SessionNotFoundError, InvalidSessionStateError } from "@fusion/dashboard/planning"; @@ -152,6 +152,7 @@ async function getBoardCommandContext(projectName?: string): Promise context.store.getSettings()}); return context; } @@ -160,13 +161,16 @@ async function getBoardCommandContext(projectName?: string): Promise context.store.getSettings()}); return context; } catch { // FNXC:PostgresCutover 2026-07-05-12:00: the cwd fallback must boot through // the PostgreSQL startup factory (createLocalStore); a bare `new TaskStore` // resolves to the removed SQLite runtime, which throws on first DB access. const store = await createLocalStore(process.cwd()); - return asLocalProjectContext(store); + const context = asLocalProjectContext(store); + installBaselineArchiveWorktreeDisposer(store, {rootDir: context.projectPath, getSettings: () => store.getSettings()}); + return context; } } diff --git a/packages/cli/src/extension.ts b/packages/cli/src/extension.ts index 553a8a730f..9ce51b7ae2 100644 --- a/packages/cli/src/extension.ts +++ b/packages/cli/src/extension.ts @@ -53,6 +53,7 @@ import { type FinalizePlanOverride, fetchWebContent, assertNoSecretPlaintext, + installBaselineArchiveWorktreeDisposer, emitGoalRetrievalAudit, createWorkflowAuthoringTools, workflowListParams, @@ -503,6 +504,7 @@ async function getStore( await boot.shutdown().catch(() => undefined); return raced.store; } + installBaselineArchiveWorktreeDisposer(boot.taskStore, {rootDir: projectRoot, getSettings: () => boot.taskStore.getSettings()}); storeCache.set(projectRoot, { store: boot.taskStore, shutdown: boot.shutdown }); return boot.taskStore; } catch (error) { @@ -550,6 +552,8 @@ async function getStore( * The entry is external: closeCachedStores / clearHostTaskStores will not shut it down — the host owns lifecycle. */ export function setHostTaskStore(projectRoot: string, store: TaskStore): void { + // FNXC:WorkflowLifecycle 2026-07-16-10:00: Install before caching an injected host store because getStore returns cached stores without a construction pass; this preserves executor-less archive cleanup during host startup. + installBaselineArchiveWorktreeDisposer(store, {rootDir: projectRoot, getSettings: () => store.getSettings()}); const canonical = resolveProjectRoot(projectRoot); storeCache.set(canonical, { store, external: true }); storeBootInflight.delete(canonical); diff --git a/packages/core/src/__tests__/worktree-path-reservation.test.ts b/packages/core/src/__tests__/worktree-path-reservation.test.ts new file mode 100644 index 0000000000..a483219c9a --- /dev/null +++ b/packages/core/src/__tests__/worktree-path-reservation.test.ts @@ -0,0 +1,58 @@ +import {mkdtemp, rm} from "node:fs/promises"; +import {tmpdir} from "node:os"; +import {join} from "node:path"; +import {afterEach, describe, expect, it, vi} from "vitest"; +import {acquireWorktreePathReservation, readWorktreePathReservation} from "../worktree-path-reservation.js"; + +const dirs: string[] = []; +async function fixture() { + const dir = await mkdtemp(join(tmpdir(), "fusion-reservation-")); + dirs.push(dir); + return {rootDir: dir, worktreesDir: join(dir, "trees"), canonicalPath: join(dir, "trees", "pinned")}; +} +afterEach(async () => { await Promise.all(dirs.splice(0).map((dir) => rm(dir, {recursive: true, force: true}))); }); + +describe("worktree path reservation", () => { + it("excludes concurrent owners until the current owner releases", async () => { + const options = await fixture(); + const first = await acquireWorktreePathReservation(options); + let acquired = false; + const second = acquireWorktreePathReservation({...options, pollMs: 1, acquireTimeoutMs: 500}).then((handle) => { acquired = true; return handle; }); + await new Promise((resolve) => setImmediate(resolve)); + expect(acquired).toBe(false); + await first.release(); + const handle = await second; + expect(handle.previousState).toBe("free"); + await handle.release(); + }); + + it("reconciles a quarantined path before handing it to a successor", async () => { + const options = await fixture(); + const first = await acquireWorktreePathReservation(options); + await first.quarantine("remove failed"); + const reconcileQuarantined = vi.fn().mockResolvedValue(undefined); + + const next = await acquireWorktreePathReservation({...options, reconcileQuarantined}); + + expect(reconcileQuarantined).toHaveBeenCalledWith(options.canonicalPath); + expect(next.previousState).toBe("quarantined"); + await next.release(); + }); + + it("keeps a quarantined path unavailable when successor reconciliation fails", async () => { + const options = await fixture(); + const first = await acquireWorktreePathReservation(options); + await first.quarantine("remove failed"); + + await expect(acquireWorktreePathReservation({...options, reconcileQuarantined: async () => { throw new Error("still occupied"); }})).rejects.toThrow("still occupied"); + + expect((await readWorktreePathReservation(options))?.state).toBe("quarantined"); + }); + + it("times out rather than waiting forever for a live claim", async () => { + const options = await fixture(); + const first = await acquireWorktreePathReservation(options); + await expect(acquireWorktreePathReservation({...options, acquireTimeoutMs: 10, pollMs: 1})).rejects.toThrow("Timed out acquiring worktree reservation"); + await first.release(); + }); +}); diff --git a/packages/core/src/archive-worktree-disposer.ts b/packages/core/src/archive-worktree-disposer.ts new file mode 100644 index 0000000000..15a059f7e6 --- /dev/null +++ b/packages/core/src/archive-worktree-disposer.ts @@ -0,0 +1,21 @@ +import type {Task} from "./types.js"; +import type {TaskStore} from "./store.js"; +import type {WorktreePathReservation} from "./worktree-path-reservation.js"; + +/** + * FNXC:WorkflowLifecycle 2026-07-16-10:00: + * Core owns archive ordering but cannot import engine. Disposers are keyed by + * store, rather than process-global, so one project's executor never removes + * a worktree for another store. Identity-guarded teardown cannot erase a newer + * executor registration. + */ +export type ArchiveWorktreeDisposer = (task: Task, reservation: WorktreePathReservation) => Promise; +const disposers = new WeakMap(); + +export function registerArchiveWorktreeDisposer(store: TaskStore, disposer: ArchiveWorktreeDisposer): () => void { + disposers.set(store, disposer); + return () => { if (disposers.get(store) === disposer) disposers.delete(store); }; +} +export function getArchiveWorktreeDisposer(store: TaskStore): ArchiveWorktreeDisposer | undefined { + return disposers.get(store); +} diff --git a/packages/core/src/index.gate.ts b/packages/core/src/index.gate.ts index d00dc75181..312562653e 100644 --- a/packages/core/src/index.gate.ts +++ b/packages/core/src/index.gate.ts @@ -528,6 +528,19 @@ export { getCreateInteractiveAiSessionFactory, type AgentMessage, } from "./ai-engine-loader.js"; +export { + registerArchiveWorktreeDisposer, + getArchiveWorktreeDisposer, + type ArchiveWorktreeDisposer, +} from "./archive-worktree-disposer.js"; +export { + acquireWorktreePathReservation, + withWorktreePathReservation, + readWorktreePathReservation, + canonicalizeWorktreePath, + type WorktreePathReservation, + type WorktreePathReservationOptions, +} from "./worktree-path-reservation.js"; export { setRunningAgentCountSource, getRunningAgentCountSource, diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 6b7fdd52d7..33aa2981ae 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -543,6 +543,19 @@ export { getCreateInteractiveAiSessionFactory, type AgentMessage, } from "./ai-engine-loader.js"; +export { + registerArchiveWorktreeDisposer, + getArchiveWorktreeDisposer, + type ArchiveWorktreeDisposer, +} from "./archive-worktree-disposer.js"; +export { + acquireWorktreePathReservation, + withWorktreePathReservation, + readWorktreePathReservation, + canonicalizeWorktreePath, + type WorktreePathReservation, + type WorktreePathReservationOptions, +} from "./worktree-path-reservation.js"; export { setRunningAgentCountSource, getRunningAgentCountSource, diff --git a/packages/core/src/task-store/archive-lifecycle-2.ts b/packages/core/src/task-store/archive-lifecycle-2.ts index cb2c7d3284..0ea954b3b3 100644 --- a/packages/core/src/task-store/archive-lifecycle-2.ts +++ b/packages/core/src/task-store/archive-lifecycle-2.ts @@ -22,6 +22,7 @@ import {softDeleteTaskRowInTransaction, readTaskRow as readTaskRowAsync} from ". import {findLiveLineageChildren as findLiveLineageChildrenAsync, projectPartition, removeLineageReferences} from "../task-store/async-lifecycle.js"; import {archiveParentTaskWithLineageGate, findArchivedTaskEntry, deleteArchivedTaskEntry, restoreTaskFromArchive} from "../task-store/async-archive-lineage.js"; import {getArchivedRowCount, listArchivedTaskEntriesPage} from "../async-archive-db.js"; +import {disposeArchivedWorktree} from "./archive-lifecycle.js"; export async function taskToArchiveEntryImpl(store: TaskStore, task: Task, archivedAt: string): Promise { const settings = await store.getSettingsFast(); @@ -190,6 +191,13 @@ export async function archiveTaskBackendImpl(store: TaskStore, id: string, optio // File-system cleanup if requested. const dir = store.taskDir(id); if (cleanup) { + /* + FNXC:WorkflowLifecycle 2026-07-16-10:00: + PostgreSQL must accept the lineage-child gate before destructive cleanup. + A rejected archive leaves its live task and pinned worktree untouched; + successful archives still await disposal before publishing the move event. + */ + await disposeArchivedWorktree(store, task); await store.cleanupBranchForTask(task); const { rm } = await import("node:fs/promises"); await rm(dir, { recursive: true, force: true }); diff --git a/packages/core/src/task-store/archive-lifecycle.ts b/packages/core/src/task-store/archive-lifecycle.ts index bc572d890e..8307ea22f8 100644 --- a/packages/core/src/task-store/archive-lifecycle.ts +++ b/packages/core/src/task-store/archive-lifecycle.ts @@ -14,6 +14,36 @@ import "../builtin-traits.js"; import {__setTaskActivityLogLimitsForTesting} from "../task-store/comments.js"; import {toJson} from "../db-helpers.js"; import {getErrorMessage} from "../error-message.js"; +import {getArchiveWorktreeDisposer} from "../archive-worktree-disposer.js"; +import {acquireWorktreePathReservation, canonicalizeWorktreePath} from "../worktree-path-reservation.js"; +import {basename, join, resolve} from "node:path"; +import {homedir} from "node:os"; + +function resolveArchiveWorktreesDir(store: TaskStore, configured?: string): string { + const value = configured?.replace(/^~(?=$|[\\/])/, homedir()).replaceAll("{repo}", basename(store.rootDir)); + return value ? resolve(store.rootDir, value) : join(store.rootDir, ".worktrees"); +} + +export async function disposeArchivedWorktree(store: TaskStore, task: Task): Promise { + if (!task.worktree) return; + const settings = await store.getSettings(); + const canonical = await canonicalizeWorktreePath(task.worktree); + if (canonical === await canonicalizeWorktreePath(store.rootDir)) return; + const reservation = await acquireWorktreePathReservation({canonicalPath: canonical, worktreesDir: resolveArchiveWorktreesDir(store, settings.worktreesDir), rootDir: store.rootDir}); + try { + const disposer = getArchiveWorktreeDisposer(store); + if (!disposer) { + /* FNXC:WorkflowLifecycle 2026-07-16-10:00: A non-root archived worktree without a store-scoped engine disposer must be loud rather than silently leaked by an executor-less archive surface. */ + storeLog.warn("archive-worktree-disposer-missing", {taskId: task.id, worktreePath: canonical}); + return; + } + try { await disposer(task, reservation); } + catch (error) { + await reservation.quarantine(getErrorMessage(error)); + storeLog.warn("Archive worktree disposal failed; reservation quarantined", {taskId: task.id, worktreePath: canonical, error: getErrorMessage(error)}); + } + } finally { if (reservation.state === "held") await reservation.release(); } +} function scheduleDeleteBranchCleanup(store: TaskStore, task: Task): void { /* @@ -248,6 +278,13 @@ export async function archiveTaskImpl(store: TaskStore, id: string, optionsOrCle return task; } + /* + FNXC:WorkflowLifecycle 2026-07-16-10:00: + Pinned paths must be reserved before destructive archive cleanup and held + until the awaited engine disposer finishes. The disposer is store-scoped + so executor-less fn/CLI archives cannot silently leak a worktree. + */ + await disposeArchivedWorktree(store, task); const cleanedBranches = await store.cleanupBranchForTask(task); if (cleanedBranches.length > 0) { task.log.push({ diff --git a/packages/core/src/worktree-path-reservation.ts b/packages/core/src/worktree-path-reservation.ts new file mode 100644 index 0000000000..26ad152897 --- /dev/null +++ b/packages/core/src/worktree-path-reservation.ts @@ -0,0 +1,149 @@ +import {createHash, randomUUID} from "node:crypto"; +import {hostname} from "node:os"; +import {dirname, join, resolve} from "node:path"; +import {mkdir, readFile, rename, rm, rmdir, unlink, writeFile} from "node:fs/promises"; + +/** + * FNXC:WorkflowLifecycle 2026-07-16-10:00: + * Pinned worktrees are reused by independent engine processes, so archive and + * acquisition share a host-scoped filesystem reservation. `claim/` is the + * exclusive mkdir create-or-fail primitive; the durable state record is never + * used as a lock because rename-over-destination does not exclude contenders. + */ +export type WorktreeReservationState = "held" | "released" | "quarantined"; +export interface WorktreePathReservation { + canonicalPath: string; + token: string; + previousState: "free" | "quarantined"; + state: WorktreeReservationState; + release(): Promise; + quarantine(reason: string): Promise; +} +interface ReservationRecord { + pid: number; + hostname: string; + startedAt: string; + canonicalPath: string; + state: "held" | "quarantined"; + token: string; + reason?: string; +} +export interface WorktreePathReservationOptions { + canonicalPath: string; + worktreesDir: string; + rootDir: string; + isLiveWorktree?: (canonicalPath: string) => Promise; + reconcileQuarantined?: (canonicalPath: string) => Promise; + ttlMs?: number; + pollMs?: number; + acquireTimeoutMs?: number; +} + +export async function canonicalizeWorktreePath(path: string): Promise { + // resolve is intentional: absent orphan paths still need one stable lock key. + return resolve(path); +} + +function paths(worktreesDir: string, canonicalPath: string) { + const key = createHash("sha256").update(canonicalPath).digest("hex"); + const container = join(resolve(worktreesDir), ".fusion-worktree-locks", key); + return {container, claim: join(container, "claim"), state: join(container, "state.json")}; +} + +async function readRecord(statePath: string): Promise { + try { + const value = JSON.parse(await readFile(statePath, "utf8")) as ReservationRecord; + return value?.token && value?.canonicalPath ? value : null; + } catch { return null; } +} +async function writeRecord(statePath: string, record: ReservationRecord): Promise { + const temp = join(dirname(statePath), `.state-${process.pid}-${randomUUID()}.tmp`); + await writeFile(temp, JSON.stringify(record), "utf8"); + await rename(temp, statePath); +} +function pidAlive(pid: number): boolean { + if (!Number.isInteger(pid) || pid <= 0) return false; + try { process.kill(pid, 0); return true; } catch (error: unknown) { + return (error as NodeJS.ErrnoException).code === "EPERM"; + } +} +const sleep = (ms: number) => new Promise((done) => setTimeout(done, ms)); + +/** Read the durable record; it is diagnostic only and does not imply ownership. */ +export async function readWorktreePathReservation(options: Pick): Promise { + const canonicalPath = await canonicalizeWorktreePath(options.canonicalPath); + return readRecord(paths(options.worktreesDir, canonicalPath).state); +} + +export async function acquireWorktreePathReservation(options: WorktreePathReservationOptions): Promise { + const canonicalPath = await canonicalizeWorktreePath(options.canonicalPath); + const layout = paths(options.worktreesDir, canonicalPath); + const pollMs = options.pollMs ?? 25; + const timeoutMs = options.acquireTimeoutMs ?? 30_000; + const ttlMs = options.ttlMs ?? 10 * 60_000; + const started = Date.now(); + await mkdir(layout.container, {recursive: true}); + + for (;;) { + try { + await mkdir(layout.claim); + const prior = await readRecord(layout.state); + const token = randomUUID(); + const record: ReservationRecord = {pid: process.pid, hostname: hostname(), startedAt: new Date().toISOString(), canonicalPath, state: "held", token}; + await writeRecord(layout.state, record); + let state: WorktreeReservationState = "held"; + const settle = async (quarantineReason?: string): Promise => { + if (state !== "held") return; + const current = await readRecord(layout.state); + if (!current || current.token !== token) { state = "released"; return; } + if (quarantineReason !== undefined) { + await writeRecord(layout.state, {...current, state: "quarantined", reason: quarantineReason}); + await rmdir(layout.claim).catch((error: unknown) => { if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; }); + state = "quarantined"; + } else { + await unlink(layout.state).catch((error: unknown) => { if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; }); + await rmdir(layout.claim).catch((error: unknown) => { if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; }); + state = "released"; + } + }; + const previousState = prior?.state === "quarantined" || prior?.state === "held" ? "quarantined" : "free"; + if (previousState === "quarantined" && options.reconcileQuarantined) { + /* + FNXC:WorkflowLifecycle 2026-07-16-10:00: + A failed archive removal is retried by the successor while its exclusive + claim is held. Re-quarantine on failure so it cannot attempt creation at + the still-occupied pinned path. + */ + try { + await options.reconcileQuarantined(canonicalPath); + } catch (error) { + await settle(error instanceof Error ? error.message : String(error)); + throw error; + } + } + return {canonicalPath, token, previousState, get state() { return state; }, release: () => settle(), quarantine: (reason) => settle(reason)}; + } catch (error: unknown) { + if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; + const record = await readRecord(layout.state); + const age = record ? Date.now() - Date.parse(record.startedAt) : 0; + const deadLocalOwner = !!record && record.hostname === hostname() && !pidAlive(record.pid); + let stale = deadLocalOwner; + if (!stale && record && age > ttlMs) { + // Fail closed: inability to prove non-liveness never steals a claim. + try { stale = !(await (options.isLiveWorktree?.(canonicalPath) ?? Promise.resolve(true))); } catch { stale = false; } + } + if (stale) { + const reclaimed = join(layout.container, `.reclaim-${randomUUID()}`); + try { await rename(layout.claim, reclaimed); await rm(reclaimed, {recursive: true, force: true}); continue; } + catch (reclaimError: unknown) { if ((reclaimError as NodeJS.ErrnoException).code !== "ENOENT") throw reclaimError; continue; } + } + if (Date.now() - started >= timeoutMs) throw new Error(`Timed out acquiring worktree reservation for ${canonicalPath} after ${timeoutMs}ms`); + await sleep(pollMs); + } + } +} + +export async function withWorktreePathReservation(options: WorktreePathReservationOptions, fn: (reservation: WorktreePathReservation) => Promise): Promise { + const reservation = await acquireWorktreePathReservation(options); + try { return await fn(reservation); } finally { if (reservation.state === "held") await reservation.release(); } +} diff --git a/packages/engine/src/archive-worktree-disposer-install.ts b/packages/engine/src/archive-worktree-disposer-install.ts new file mode 100644 index 0000000000..33e5801940 --- /dev/null +++ b/packages/engine/src/archive-worktree-disposer-install.ts @@ -0,0 +1,18 @@ +import {canonicalizeWorktreePath, getArchiveWorktreeDisposer, registerArchiveWorktreeDisposer, type Settings, type TaskStore} from "@fusion/core"; +import {removeWorktree, RemovalReason} from "./worktree-backend.js"; + +/** + * FNXC:WorkflowLifecycle 2026-07-16-10:00: + * CLI/fn archive paths can own a store without constructing an executor. This + * presence-guarded baseline uses the configured backend, while an executor may + * replace it with its session-aware disposer for the same store. + */ +export function installBaselineArchiveWorktreeDisposer(store: TaskStore, input: {rootDir: string; getSettings: () => Promise>}): () => void { + if (getArchiveWorktreeDisposer(store)) return () => {}; + return registerArchiveWorktreeDisposer(store, async (task) => { + if (!task.worktree) return; + if (await canonicalizeWorktreePath(task.worktree) === await canonicalizeWorktreePath(input.rootDir)) return; + await removeWorktree({worktreePath: task.worktree, rootDir: input.rootDir, settings: await input.getSettings(), taskId: task.id, reason: RemovalReason.ExecutorDispose, force: true}); + task.worktree = undefined; + }); +} diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 60f370ad54..4c4c3469c3 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -135,6 +135,7 @@ import { deriveRepoScopeSubset, normalizeRepoRelPath } from "./workspace-paths.j import { RemovalReason, classifyTaskWorktree, describeRegisteredWorktrees, detectGitRepository, detectNestedWorktreeRoot, getRegisteredWorktreePaths, isInsideWorktreesDir, isRegisteredGitWorktree, removeWorktree, type GitRepoDetection, type WorktreePool } from "./worktree-pool.js"; import { attemptBranchAutocorrect } from "./branch-autocorrect.js"; import { ActiveSessionWorktreeRemovalError } from "./worktree-backend.js"; +import {canonicalizeWorktreePath, registerArchiveWorktreeDisposer} from "@fusion/core"; import { activeSessionRegistry, executingTaskLock, @@ -1710,6 +1711,7 @@ export class TaskExecutor { * this so a fast re-dispatch (task:moved → in-progress) awaits the prior * session being fully reaped before creating/acquiring a new worktree. */ private pendingTaskDisposals = new Map>(); + private unregisterArchiveWorktreeDisposer: (() => void) | undefined; /** Active agent sessions per task, used to terminate on pause and inject steering. */ private activeSessions = new Map(); /** Active step-session executors per task (mutually exclusive with activeSessions). */ @@ -2956,6 +2958,14 @@ export class TaskExecutor { private options: TaskExecutorOptions = {}, ) { executorLog.log(`TaskExecutor constructed (rootDir=${rootDir}, hasSemaphore=${!!options.semaphore}, hasStuckDetector=${!!options.stuckTaskDetector})`); + /* FNXC:WorkflowLifecycle 2026-07-16-10:00: Executor replaces the baseline only for its own TaskStore, so archive awaits abort/sweep/removal before branch deletion without cross-store coupling. */ + this.unregisterArchiveWorktreeDisposer = registerArchiveWorktreeDisposer(store, async (task) => { + if (!task.worktree || await canonicalizeWorktreePath(task.worktree) === await canonicalizeWorktreePath(this.rootDir)) return; + await this.awaitAbortInFlightTaskWork(task.id, "task archived"); + for (const path of activeSessionRegistry.pathsForTask(task.id)) activeSessionRegistry.unregisterPath(path); + await this.removeOwnWorktreeWithReconcile({worktreePath: task.worktree, settings: await store.getSettings(), taskId: task.id, reason: RemovalReason.ExecutorDispose}); + task.worktree = undefined; + }); store.on("task:moved", ({ task, from, to, source }) => { executorLog.log(`[event:task:moved] ${task.id}: ${from} → ${to}`); @@ -18109,6 +18119,12 @@ You have access to the file system to review changes.${inlineFixBlock}${verdictB } } + /** Remove only this executor's store-scoped archive disposer registration. */ + disposeArchiveWorktreeDisposer(): void { + this.unregisterArchiveWorktreeDisposer?.(); + this.unregisterArchiveWorktreeDisposer = undefined; + } + private async removeOwnWorktreeWithReconcile(input: { worktreePath: string; settings: Settings; diff --git a/packages/engine/src/index.ts b/packages/engine/src/index.ts index b0dba24c06..5451c7298b 100644 --- a/packages/engine/src/index.ts +++ b/packages/engine/src/index.ts @@ -1136,6 +1136,8 @@ export { genericCliAdapter, type CliAdapterDescriptor, } from "./cli-agent/adapters/index.js"; +export { installBaselineArchiveWorktreeDisposer } from "./archive-worktree-disposer-install.js"; + // CLI Agent Executor — task ↔ session orchestration (U7). export { CliTaskSession, diff --git a/packages/engine/src/runtimes/in-process-runtime.ts b/packages/engine/src/runtimes/in-process-runtime.ts index 269440d5bb..bf97036008 100644 --- a/packages/engine/src/runtimes/in-process-runtime.ts +++ b/packages/engine/src/runtimes/in-process-runtime.ts @@ -1387,6 +1387,7 @@ export class InProcessRuntime if (this.executor) { try { await this.executor.abortAllInFlight("engine stop"); + this.executor.disposeArchiveWorktreeDisposer(); runtimeLog.log("Aborted in-flight executor AI sessions"); } catch (err) { runtimeLog.warn(`Failed to abort in-flight executor AI sessions: ${err}`); diff --git a/packages/engine/src/worktree-acquisition.ts b/packages/engine/src/worktree-acquisition.ts index 575c9f9010..51711a4962 100644 --- a/packages/engine/src/worktree-acquisition.ts +++ b/packages/engine/src/worktree-acquisition.ts @@ -1,9 +1,9 @@ import { existsSync } from "node:fs"; import { exec } from "node:child_process"; import { promisify } from "node:util"; -import type { RunMutationContext, Settings, Task, TaskStore, SecretsStore } from "@fusion/core"; +import {acquireWorktreePathReservation, canonicalizeWorktreePath, type RunMutationContext, type Settings, type Task, type TaskStore, type SecretsStore} from "@fusion/core"; import { generateWorktreeName, resolveTaskWorkingBranch, slugify } from "./worktree-names.js"; -import { resolveTaskWorktreePathForBackend } from "./worktree-paths.js"; +import { resolveTaskWorktreePathForBackend, resolveWorktreesDir } from "./worktree-paths.js"; import { hydrateWorktreeDb } from "./worktree-db-hydrate.js"; import { formatError } from "./logger.js"; import { classifyBootstrapMisbinding, isBranchConflictError, reanchorBranchToBase } from "./branch-conflicts.js"; @@ -283,6 +283,27 @@ export async function acquireTaskWorktree(opts: AcquireTaskWorktreeOptions): Pro const createWorktreeImpl = createWorktree ? createWorktree : async (createBranch: string, createPath: string, createTaskId: string, startPoint?: string, allowRename?: boolean) => { + const reservation = await acquireWorktreePathReservation({ + canonicalPath: await canonicalizeWorktreePath(createPath), + worktreesDir: resolveWorktreesDir(rootDir, settings), + rootDir, + /* + FNXC:WorkflowLifecycle 2026-07-16-10:00: + A failed archive removal leaves a durable quarantine record. The next + owner must reconcile that old pinned path while it exclusively holds + the reservation, rather than colliding with it during creation. + */ + reconcileQuarantined: async () => { + await removeWorktree({ + worktreePath: createPath, + rootDir, + settings, + taskId: createTaskId, + reason: RemovalReason.ExecutorDispose, + force: true, + }); + }, + }); try { const created = await backend.create({ rootDir, @@ -315,6 +336,8 @@ export async function acquireTaskWorktree(opts: AcquireTaskWorktreeOptions): Pro return await handleWorktrunkFailure("create", error, fallback) as { path: string; branch: string }; } throw error; + } finally { + if (reservation.state === "held") await reservation.release(); } };