diff --git a/.changeset/fn-9052-workspace-worktree-atomic-merge.md b/.changeset/fn-9052-workspace-worktree-atomic-merge.md new file mode 100644 index 0000000000..c0f8c8d5b9 --- /dev/null +++ b/.changeset/fn-9052-workspace-worktree-atomic-merge.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Fix lost sub-repo worktree entries when workspace repos are acquired concurrently. +category: fix +dev: Per-repo workspace state now uses mergeWorkspaceWorktreeEntry under the task advisory lock. diff --git a/docs/architecture.md b/docs/architecture.md index 632bc4cb27..b84c7e7279 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -627,6 +627,7 @@ See [Memory Plugin Contract](./memory-plugin-contract.md) for the full plan. - **Planning**: the planning processor generates task plans (`PROMPT.md`) and selects eligible planning tasks by priority first, then FIFO (`createdAt` ascending) within each priority tier. Each attempt captures the authoritative artifact baseline and owns its fallback callback provenance. Only a settled, fallback-free attempt that changed that exact baseline and passes deterministic validation may hand off to workflow Plan Review. Empty, unchanged, or fallback-engaged attempts use the shared bounded `recoveryRetryCount`/`nextRecoveryAt` backoff; exhaustion persists an actionable planning error and never signals successful handoff. After a prompt settles, triage awaits the originating runtime's finite `settleFallbackDispatch` lifecycle signal, then awaits every observer callback admitted by that signal before deciding. A configured runtime that cannot supply this signal fails closed through the same bounded planning recovery rather than handing a potentially fallback-authored plan to review. This deliberately never inspects arbitrary Node timers: clean planner housekeeping can schedule unrelated one-shot or recurring timers without delaying admission. A callback from an obsolete attempt remains scoped to that attempt. Explicit duplicate-marker closure runs only after this same clean-attempt admission. If the stuck-task detector kills a not-yet-approved planning session after a non-empty `PROMPT.md` draft exists, the retry is requeued as `needs-replan` and seeds the next prompt in revision mode from that draft instead of cold-starting. A newly added dependency in a hold lane follows the same durable `needs-replan` path; it never clears status, so a planner interrupted after prompt persistence remains claimable and cannot silently bypass the approval/release handoff. When `PROMPT.md` is absent, a non-empty `plan` task document written through `fn_task_document_write` is the fallback seed; missing or whitespace-only drafts still cold-start. - **Executor**: `TaskExecutor` (`executor.ts`) implements tasks in worktrees - **Workspace acquisition shape:** per-repo acquisition persists only `workspaceWorktrees`; it never exposes a sub-repo path or branch through singular `task.worktree`/`task.branch`, including if the final workspace-state write fails. This preserves workspace classification for dashboard rendering, self-healing, and executor dispatch. + - **Workspace entry mutation (FN-9052):** every per-repository `workspaceWorktrees` update goes through `TaskStore.mergeWorkspaceWorktreeEntry`, which holds the per-task PostgreSQL advisory transaction lock and merges one key under the composite project/task scope. Per-repo callers must never wholesale-replace the map, because a concurrent sibling acquisition, landing, failure, or teardown mutation would otherwise lose its entry. - **Main-checkout completion guard (FN-9058):** `fn_task_done` probes every configured sub-repo main checkout before any workspace worktree invariant, so `main_checkout_edit` takes precedence over `no_commits` and cannot be skipped by zero-acquire or no-commit eligibility. It uses the immutable first-execution anchor (never only the re-stamped per-attempt timestamp), blocks task-era status entries and bounded recent-HEAD evidence without `--since` or ancestry filtering, and emits `worktree:workspace-main-checkout-edit`. Unattributable pre-existing dirt, unavailable probes, and unresolved timing only warn: refusal has a bounded requeue budget and the guard is read-only. - **Task-pinned orphan recovery:** task-ID-pinned acquisition holds one path reservation across classification, preservation, quarantine reconciliation, and recreation. Inactive incomplete or unregistered directories are atomically moved to `/.fusion/recovery/worktrees`, or to `/.fusion-recovery/worktrees` after an `EXDEV` cross-filesystem refusal. Each actual recovery root retains the newest 10 recognized Fusion-generated entries; pruning is fail-soft and preserves unknown, symlinked, unreadable, or active paths. Worktree pool and self-healing scans exclude both `.ai-merge` and `.fusion-recovery` as internal container boundaries. diff --git a/packages/core/src/__tests__/postgres/workspace-worktrees-concurrent-merge.pg.test.ts b/packages/core/src/__tests__/postgres/workspace-worktrees-concurrent-merge.pg.test.ts new file mode 100644 index 0000000000..c157031a67 --- /dev/null +++ b/packages/core/src/__tests__/postgres/workspace-worktrees-concurrent-merge.pg.test.ts @@ -0,0 +1,90 @@ +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from "vitest"; +import { createSharedPgTaskStoreTestHarness, pgDescribe, type SharedPgTaskStoreHarness } from "../../__test-utils__/pg-test-harness.js"; +import { isWorkspaceTask } from "../../types.js"; + +const pgTest = pgDescribe; + +/* +FNXC:Workspace 2026-08-15-07:51: +These tests use distinct TaskStore handles against one PostgreSQL database. An in-process mutex +would make the concurrent cases pass accidentally; only the task advisory transaction lock makes +both independently issued per-repo merges retain their sibling entries. +*/ +pgTest("workspace worktree per-repo atomic merge (PostgreSQL)", () => { + const h: SharedPgTaskStoreHarness = createSharedPgTaskStoreTestHarness({ prefix: "fusion_workspace_merge" }); + + beforeAll(h.beforeAll); + beforeEach(h.beforeEach); + afterEach(h.afterEach); + afterAll(h.afterAll); + + it("retains both different repo keys from genuinely concurrent store handles", async () => { + const first = h.store(); + const second = h.store(); + const task = await first.createTask({ description: "concurrent workspace merges" }); + + await Promise.all([ + first.mergeWorkspaceWorktreeEntry(task.id, "repo-a", { worktreePath: "/tmp/repo-a", branch: "fusion/a", baseCommitSha: "base-a" }), + second.mergeWorkspaceWorktreeEntry(task.id, "repo-b", { worktreePath: "/tmp/repo-b", branch: "fusion/b", baseCommitSha: "base-b" }), + ]); + + const current = await first.getTask(task.id); + expect(current.workspaceWorktrees).toEqual({ + "repo-a": { worktreePath: "/tmp/repo-a", branch: "fusion/a", baseCommitSha: "base-a" }, + "repo-b": { worktreePath: "/tmp/repo-b", branch: "fusion/b", baseCommitSha: "base-b" }, + }); + }); + + it("preserves an existing entry while a sibling is added concurrently", async () => { + const first = h.store(); + const second = h.store(); + const task = await first.createTask({ description: "landed SHA and acquisition" }); + await first.mergeWorkspaceWorktreeEntry(task.id, "repo-a", { + worktreePath: "/tmp/repo-a", branch: "fusion/a", baseCommitSha: "base-a", + }); + + await Promise.all([ + first.mergeWorkspaceWorktreeEntry(task.id, "repo-a", { landedSha: "landed-a" }, { requireExistingEntry: true }), + second.mergeWorkspaceWorktreeEntry(task.id, "repo-b", { worktreePath: "/tmp/repo-b", branch: "fusion/b" }), + ]); + + expect((await first.getTask(task.id)).workspaceWorktrees).toEqual({ + "repo-a": { worktreePath: "/tmp/repo-a", branch: "fusion/a", baseCommitSha: "base-a", landedSha: "landed-a" }, + "repo-b": { worktreePath: "/tmp/repo-b", branch: "fusion/b" }, + }); + }); + + it("clears singular state in the same per-key update", async () => { + const store = h.store(); + const task = await store.createTask({ description: "workspace singular state" }); + await store.updateTask(task.id, { worktree: "/tmp/legacy", branch: "fusion/legacy" }); + + const updated = await store.mergeWorkspaceWorktreeEntry(task.id, "repo-a", { + worktreePath: "/tmp/repo-a", branch: "fusion/a", + }, { clearSingularWorktree: true }); + + expect(updated.worktree).toBeUndefined(); + expect(updated.branch).toBeUndefined(); + expect(isWorkspaceTask(updated)).toBe(true); + }); + + it("does not create an absent required entry or clobber siblings", async () => { + const store = h.store(); + const task = await store.createTask({ description: "required entry no-op" }); + await store.mergeWorkspaceWorktreeEntry(task.id, "repo-a", { worktreePath: "/tmp/repo-a", branch: "fusion/a" }); + + const unchanged = await store.mergeWorkspaceWorktreeEntry(task.id, "repo-b", { landedSha: "ignored" }, { requireExistingEntry: true }); + expect(unchanged.workspaceWorktrees).toEqual({ "repo-a": { worktreePath: "/tmp/repo-a", branch: "fusion/a" } }); + }); + + it.each([undefined, {}] as const)("creates one entry from %j workspace state", async (workspaceWorktrees) => { + const store = h.store(); + const task = await store.createTask({ description: "empty workspace state" }); + if (workspaceWorktrees) await store.updateTask(task.id, { workspaceWorktrees }); + + const updated = await store.mergeWorkspaceWorktreeEntry(task.id, "repo-a", { worktreePath: "/tmp/repo-a", branch: "fusion/a" }); + expect(updated.workspaceWorktrees).toEqual({ "repo-a": { worktreePath: "/tmp/repo-a", branch: "fusion/a" } }); + }); +}); + +void describe; diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index ba557b6a26..42d0401b80 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -123,7 +123,7 @@ import { getTaskCommitAssociationsByLineageIdImpl, replaceLegacyTaskCommitAssoci import { findRecentTasksBySourceParentTaskIdImpl } from "./task-store/branch-and-pr-entities.js"; import { addTaskCommentImpl, applyBuiltInPromptOverridesAsyncImpl, applyBuiltInPromptOverridesSyncImpl, areAllDependenciesDoneImpl, artifactStoredNameImpl, assertWorkflowIrTraitsValidImpl, clearActivityLogImpl, clearTaskWorkflowSelectionImpl, deleteTaskByIdImpl, getDefaultWorkflowIdImpl, resolveOriginWorkflowOverrideIdImpl, type TaskOriginWorkflowKind, getInsightStoreImpl, getMergeQueuedTaskIdsImpl, getMergeRequestRecordImpl, getMergeRequestRecordAsyncImpl, getResearchStoreImpl, getTaskIdFromDirImpl, getTodoStoreImpl, getWorkflowWorkItemByIdentityImpl, hasActiveTaskImpl, invalidateConfigCacheAfterMigrationImpl, isTaskIdConflictErrorImpl, listLegacyAutoMergeStampCandidatesImpl, readTaskRowFromDbImpl, recordBranchGroupMemberLandedImpl, refreshDatabaseHealthAsyncImpl, refreshDatabaseHealthImpl, resolveTaskCustomFieldDefsSyncImpl, resolveWorkflowBypassGuardsImpl, serializeConfigForDiskImpl, setPluginWorkflowStepTemplatesImpl, shouldSkipWorkflowMovePoliciesImpl, suppressWatcherImpl, upsertTaskWithFtsRecoveryImpl } from "./task-store/task-store-helpers.js"; import { getTaskSelectClauseImpl2, createTaskPersistSerializationContextImpl, getTaskPersistValuesImpl, getTaskPatchDescriptorsImpl, normalizeTaskFromDiskImpl, writeTaskJsonFileImpl, rowToPrEntityImpl, generatePrEntityIdImpl, readTaskForMoveImpl, rowToMergeQueueEntryImpl, rowToMergeRequestRecordImpl, rowToCompletionHandoffMarkerImpl, rowToWorkflowWorkItemImpl, rowToRunAuditEventImpl } from "./task-store/task-row-mappers.js"; -import { getTaskSelectClauseWithActivityLogLimitImpl, getChangedTaskColumnsImpl, getSoftDeletedWriteConflictImpl, readTaskJsonImpl, writeConfigImpl, _maybeAutoArchiveSameAgentDuplicateBackendImpl, updateBranchGroupImpl, updatePrEntityImpl, listTasksForGithubTrackingReconcileImpl, listTasksForGitlabTrackingReconcileImpl, renewCheckoutLeaseImpl, updateTaskAtomicImpl, linkTaskRecommendationImpl, resolveTaskWedgeNotificationEpisodeImpl, getWorkflowPromptOverridesImpl, updateWorkflowSettingValuesImpl, rollbackConfigurationImpl, cancelActiveWorkflowWorkItemsForTaskImpl, setCompletionHandoffAcceptedMarkerImpl, reconcileLegacyAutoMergeStampsImpl, recoverExpiredMergeQueueLeasesImpl, rewriteDependentsForRemovalImpl, cleanupBranchForTaskImpl, addAttachmentImpl, deleteAttachmentImpl, registerArtifactImpl, updatePrInfoImpl, unlinkGithubIssueImpl, cleanupArchivedTasksImpl, generatePromptFromArchiveEntryImpl, listWorkflowOccupantTaskIdsImpl, listApprovedCliAutonomyAdaptersImpl, closeImpl, getActivityLogImpl } from "./task-store/task-mutation-ops.js"; +import { getTaskSelectClauseWithActivityLogLimitImpl, getChangedTaskColumnsImpl, getSoftDeletedWriteConflictImpl, readTaskJsonImpl, writeConfigImpl, _maybeAutoArchiveSameAgentDuplicateBackendImpl, updateBranchGroupImpl, updatePrEntityImpl, listTasksForGithubTrackingReconcileImpl, listTasksForGitlabTrackingReconcileImpl, renewCheckoutLeaseImpl, updateTaskAtomicImpl, linkTaskRecommendationImpl, mergeWorkspaceWorktreeEntryImpl, resolveTaskWedgeNotificationEpisodeImpl, getWorkflowPromptOverridesImpl, updateWorkflowSettingValuesImpl, rollbackConfigurationImpl, cancelActiveWorkflowWorkItemsForTaskImpl, setCompletionHandoffAcceptedMarkerImpl, reconcileLegacyAutoMergeStampsImpl, recoverExpiredMergeQueueLeasesImpl, rewriteDependentsForRemovalImpl, cleanupBranchForTaskImpl, addAttachmentImpl, deleteAttachmentImpl, registerArtifactImpl, updatePrInfoImpl, unlinkGithubIssueImpl, cleanupArchivedTasksImpl, generatePromptFromArchiveEntryImpl, listWorkflowOccupantTaskIdsImpl, listApprovedCliAutonomyAdaptersImpl, closeImpl, getActivityLogImpl } from "./task-store/task-mutation-ops.js"; import { getOrCreateForProjectImpl, listGoalCitationsImpl, atomicWriteTaskJsonWithAuditImpl, type PlanningDependencyInvalidation, duplicateTaskImpl, listStrandedRefinementsImpl, tryClaimCheckoutImpl, evaluateWorkflowMovePoliciesImpl, recordRunAuditEventImpl, getRunAuditEventsImpl, dequeueMergeQueueOnColumnExitImpl, updateIssueInfoImpl, listWorkflowStepsImpl, getWorkflowStepImpl, createWorkflowDefinitionImpl, countActiveInCapacitySlotSyncImpl, countActiveInCapacitySlotAsyncImpl, generateSpecifiedPromptImpl, recordActivityImpl, getEvalStoreImpl } from "./task-store/project-store-ops.js"; import { markLegacyAutoMergeStampsOnceImpl, appendAgentLogImpl, importLegacyAgentLogsImpl, cleanupNoOpTaskMovedActivityRowsOnceImpl, backfillCommitAssociationDiffStatsImpl } from "./task-store/workflow-integrity.js"; import { saveWorkflowRunBranchImpl, clearNearDuplicateReferencesToImpl, selectNextTaskForAgentImpl, pauseTaskImpl, clearLinkedAgentTaskIdsImpl, listArtifactsImpl, rehomeOccupantImpl, type RehomeOccupantResult } from "./task-store/branch-group-ops.js"; @@ -2015,6 +2015,14 @@ export class TaskStore extends EventEmitter { ): Promise { return linkTaskRecommendationImpl(this, id, recommendationId, createdTaskId, completeColumns); } + async mergeWorkspaceWorktreeEntry( + id: string, + repoRelPath: string, + patch: Partial, + options?: { requireExistingEntry?: boolean; clearSingularWorktree?: boolean }, + ): Promise { + return mergeWorkspaceWorktreeEntryImpl(this, id, repoRelPath, patch, options); + } async resolveTaskWedgeNotificationEpisode(id: string, episodeId: string): Promise<{ task: Task; resolved: boolean }> { return resolveTaskWedgeNotificationEpisodeImpl(this, id, episodeId); } diff --git a/packages/core/src/task-store/task-mutation-ops.ts b/packages/core/src/task-store/task-mutation-ops.ts index 461fc0aa4d..d411df8836 100644 --- a/packages/core/src/task-store/task-mutation-ops.ts +++ b/packages/core/src/task-store/task-mutation-ops.ts @@ -19,7 +19,7 @@ import {randomUUID} from "node:crypto"; import {mkdir, readFile, writeFile, rename, unlink} from "node:fs/promises"; import {join} from "node:path"; import {existsSync} from "node:fs"; -import type {Task, TaskCreateInput, TaskAttachment, BoardConfig, ActivityLogEntry, ActivityEventType, Artifact, ArtifactCreateInput, RunMutationContext, MergeQueueEntry, BranchGroup, BranchGroupUpdate, CompletionHandoffMarker, WorkflowWorkItem, WorkflowWorkItemKind, PrEntity, PrEntityUpdate, TaskRecommendation} from "../types.js"; +import type {Task, TaskCreateInput, TaskAttachment, BoardConfig, ActivityLogEntry, ActivityEventType, Artifact, ArtifactCreateInput, RunMutationContext, MergeQueueEntry, BranchGroup, BranchGroupUpdate, CompletionHandoffMarker, WorkflowWorkItem, WorkflowWorkItemKind, PrEntity, PrEntityUpdate, TaskRecommendation, WorkspaceWorktreeEntry} from "../types.js"; import { CONFIG_CHANGED_BY_SYSTEM } from "../types.js"; import {validateSettingValuePatch, WorkflowSettingRejectionError} from "../workflows/workflow-settings.js"; import "../builtin-traits.js"; @@ -457,6 +457,56 @@ Resolution changes only the active episode status. The PostgreSQL compare-and-se merges that field into the existing JSON, preserving per-reason cooldown stamps so resolving X or notifying Y cannot reopen X's live spam window. */ +/* +FNXC:Workspace 2026-08-15-07:51: +Per-repository workspace worktree updates must serialize across Fusion processes, not merely one +TaskStore instance. The advisory transaction lock is acquired before reading the composite +(project_id, id)-scoped row, then this method replaces only the requested key. Reintroducing an +engine-side wholesale workspaceWorktrees update would reopen the Phase-B sibling-clobber race. +*/ +export async function mergeWorkspaceWorktreeEntryImpl( + store: TaskStore, + id: string, + repoRelPath: string, + patch: Partial, + options: { requireExistingEntry?: boolean; clearSingularWorktree?: boolean } = {}, +): Promise { + return store.withTaskLock(id, async () => { + const layer = store.asyncLayer!; + const outcome = await layer.transactionImmediate(async (tx) => { + await acquireTaskAdvisoryXactLock(tx, layer.projectId, id); + const row = await readTaskRowInTransaction(tx, id, { includeDeleted: true }, layer.projectId); + if (!row) throw new TaskNotFoundError(id); + if (row.deletedAt) throw new TaskDeletedError(id, row.deletedAt as string); + + const current = store.rowToTask(store.pgRowToTaskRow(row)); + const workspaceWorktrees = current.workspaceWorktrees ?? {}; + const existing = workspaceWorktrees[repoRelPath]; + if (options.requireExistingEntry && !existing) return { task: current, mutated: false }; + + const updatedAt = new Date().toISOString(); + const [updatedRow] = await tx + .update(schema.project.tasks) + .set({ + workspaceWorktrees: { ...workspaceWorktrees, [repoRelPath]: { ...existing, ...patch } }, + ...(options.clearSingularWorktree ? { worktree: null, branch: null } : {}), + updatedAt, + }) + .where(and(eq(schema.project.tasks.id, id), taskProjectScope(layer))) + .returning(); + if (!updatedRow) throw new TaskNotFoundError(id); + return { task: store.rowToTask(store.pgRowToTaskRow(updatedRow)), mutated: true }; + }); + + if (outcome.mutated) { + await store.writeTaskJsonFile(store.taskDir(id), outcome.task); + if (store.isWatching) store.taskCache.set(id, { ...outcome.task }); + store.emitTaskLifecycleEventSafely("task:updated", [outcome.task]); + } + return outcome.task; + }); +} + export async function resolveTaskWedgeNotificationEpisodeImpl( store: TaskStore, id: string, diff --git a/packages/core/src/types.ts b/packages/core/src/types.ts index 14c52b0f6e..d439ce64af 100644 --- a/packages/core/src/types.ts +++ b/packages/core/src/types.ts @@ -610,6 +610,7 @@ import type { TaskRecommendationCategory, TaskRecommendationListItem, TaskRecommendationListPage, + WorkspaceWorktreeEntry, Task, TaskReleaseGateVerdict, TaskVerificationResultSummary, @@ -657,6 +658,7 @@ export type { TaskRecommendationCategory, TaskRecommendationListItem, TaskRecommendationListPage, + WorkspaceWorktreeEntry, Task, TaskReleaseGateVerdict, TaskVerificationResultSummary, diff --git a/packages/core/src/types/task/task-core.ts b/packages/core/src/types/task/task-core.ts index 7456e92b0a..5f3fc50468 100644 --- a/packages/core/src/types/task/task-core.ts +++ b/packages/core/src/types/task/task-core.ts @@ -675,6 +675,21 @@ export interface TaskReleaseGateVerdict { evaluatedForUpdatedAt?: string; } +/* +FNXC:Workspace 2026-08-15-07:51: +The atomic per-repository store mutation and its engine callers share this entry contract so +per-key merges preserve every durable workspace worktree field rather than drifting into +independent inline shapes. +*/ +export interface WorkspaceWorktreeEntry { + worktreePath: string; + branch: string; + baseCommitSha?: string; + landedSha?: string; + revertBoundarySha?: string; + landFailure?: { message: string; at: string; branch?: string }; +} + export interface Task { id: string; /** Immutable lineage identity used for durable commit/task attribution. */ @@ -732,7 +747,7 @@ export interface Task { * error prose is never parsed for attribution. FN-9047/FN-9048 stale-state clearing must drop * it alongside `landedSha`. */ - workspaceWorktrees?: Record; + workspaceWorktrees?: Record; steps: TaskStep[]; currentStep: number; /** diff --git a/packages/engine/src/__tests__/merge-orphan-body-durable-writes.test.ts b/packages/engine/src/__tests__/merge-orphan-body-durable-writes.test.ts index 2e99f4e003..38cdde6b0f 100644 --- a/packages/engine/src/__tests__/merge-orphan-body-durable-writes.test.ts +++ b/packages/engine/src/__tests__/merge-orphan-body-durable-writes.test.ts @@ -72,6 +72,14 @@ function createRecordingStore(controller: AbortController, options: { sharedGrou if (patch.mergeDetails) options.onMergeDetailsPersist?.(); return task; }), + mergeWorkspaceWorktreeEntry: vi.fn(async (_id: string, repoRelPath: string, patch: Record, mergeOptions?: { requireExistingEntry?: boolean }) => { + records.push({ generation, writer: "mergeWorkspaceWorktreeEntry", args: [_id, repoRelPath, patch, mergeOptions] }); + const current = (task.workspaceWorktrees as Record> | undefined) ?? {}; + const existing = current[repoRelPath]; + if (mergeOptions?.requireExistingEntry && !existing) return task; + task.workspaceWorktrees = { ...current, [repoRelPath]: { ...existing, ...patch } }; + return task; + }), moveTask: vi.fn(async (...args: unknown[]) => { records.push({ generation, writer: "moveTask", args }); task.column = args[1] as string; return task; }), logEntry: record("logEntry"), appendAgentLog: record("appendAgentLog"), emit: vi.fn((...args: unknown[]) => { records.push({ generation, writer: "emit", args }); }), diff --git a/packages/engine/src/__tests__/self-healing-workspace.test.ts b/packages/engine/src/__tests__/self-healing-workspace.test.ts index 1c9fbb5e35..42290f43d1 100644 --- a/packages/engine/src/__tests__/self-healing-workspace.test.ts +++ b/packages/engine/src/__tests__/self-healing-workspace.test.ts @@ -45,6 +45,7 @@ interface RecordingStore extends EventEmitter { emitted: Array<{ event: string; payload: unknown }>; enqueued: string[]; updateTask: ReturnType; + mergeWorkspaceWorktreeEntry: ReturnType; moveTask: ReturnType; } @@ -69,6 +70,25 @@ function createStore(rows: Task[], settings: Partial = {}): TaskStore if (cur) tasks.set(id, { ...cur, ...patch } as Task); return tasks.get(id) as Task; }), + mergeWorkspaceWorktreeEntry: vi.fn(async ( + id: string, + repoRel: string, + patch: Partial[string]>, + options?: { requireExistingEntry?: boolean; clearSingularWorktree?: boolean }, + ) => { + const current = tasks.get(id); + if (!current) throw new Error(`Task ${id} not found`); + const workspaceWorktrees = current.workspaceWorktrees ?? {}; + const existing = workspaceWorktrees[repoRel]; + if (options?.requireExistingEntry && !existing) return current; + const updated = { + ...current, + workspaceWorktrees: { ...workspaceWorktrees, [repoRel]: { ...existing, ...patch } }, + ...(options?.clearSingularWorktree ? { worktree: undefined, branch: undefined } : {}), + } as Task; + tasks.set(id, updated); + return updated; + }), moveTask: vi.fn(async (id: string, column: string) => { const cur = tasks.get(id); const next = { ...(cur ?? { id }), column } as Task; @@ -926,7 +946,12 @@ describeIfGit("workspace-aware self-healing (Phase D U1)", () => { expect(await manager.reconcileOrphanedWorkspaceWorktrees()).toBe(1); expect(existsSync(worktreePath)).toBe(false); expect(fx.git("repo-a", `git branch --list ${BRANCH}`).trim()).toBe(""); - expect(store.updateTask).toHaveBeenCalled(); + expect(store.mergeWorkspaceWorktreeEntry).toHaveBeenCalledWith( + TASK_ID, + "repo-a", + { worktreePath: "" }, + { requireExistingEntry: true }, + ); }); /* @@ -976,9 +1001,12 @@ describeIfGit("workspace-aware self-healing (Phase D U1)", () => { expect(await makeManager(store, fx.rootDir).reconcileOrphanedWorkspaceWorktrees()).toBe(1); expect(existsSync(worktreePath)).toBe(false); expect(fx.git("repo-a", `git branch --list ${BRANCH}`).trim()).toContain(BRANCH); - expect(store.updateTask).toHaveBeenCalledWith(TASK_ID, expect.objectContaining({ - workspaceWorktrees: expect.objectContaining({ "repo-a": expect.objectContaining({ branch: BRANCH, baseCommitSha, worktreePath: "" }) }), - })); + expect(store.mergeWorkspaceWorktreeEntry).toHaveBeenCalledWith( + TASK_ID, + "repo-a", + { worktreePath: "" }, + { requireExistingEntry: true }, + ); }); it("prunes an already-gone recorded worktree and settles its absent branch", async () => { @@ -1011,9 +1039,12 @@ describeIfGit("workspace-aware self-healing (Phase D U1)", () => { const manager = makeManager(store, fx.rootDir); expect(await manager.reconcileOrphanedWorkspaceWorktrees()).toBe(1); - expect(store.updateTask).toHaveBeenCalledWith(TASK_ID, expect.objectContaining({ - workspaceWorktrees: expect.objectContaining({ "repo-a": expect.objectContaining({ branch: BRANCH, worktreePath: "" }) }), - })); + expect(store.mergeWorkspaceWorktreeEntry).toHaveBeenCalledWith( + TASK_ID, + "repo-a", + { worktreePath: "" }, + { requireExistingEntry: true }, + ); expect(await manager.reconcileOrphanedWorkspaceWorktrees()).toBe(0); }); @@ -1091,7 +1122,7 @@ describeIfGit("workspace-aware self-healing (Phase D U1)", () => { // Negative scope: one primary veto cannot silently disable teardown of another terminal row. expect(existsSync(companionPath)).toBe(false); expect(fx.git("repo-a", `git branch --list ${companionBranch}`).trim()).toBe(""); - expect(store.updateTask).toHaveBeenCalledTimes(1); + expect(store.mergeWorkspaceWorktreeEntry).toHaveBeenCalledTimes(1); }); it("settles the prune phase while retaining a duplicate branch claim without re-pruning", async () => { @@ -1176,7 +1207,7 @@ describeIfGit("workspace-aware self-healing (Phase D U1)", () => { rmSync(worktreePath, { recursive: true, force: true }); const rejected = workspaceTask({ "repo-a": { worktreePath, branch: BRANCH, landedSha: "landed" } }, { id: "FN-7003", deletedAt: old, updatedAt: old, columnMovedAt: old }); const store = createStore([rejected]); - store.updateTask.mockRejectedValue(new Error("soft-deleted")); + store.mergeWorkspaceWorktreeEntry.mockRejectedValue(new Error("soft-deleted")); const settled = makeManager(store, fx.rootDir); expect(await settled.reconcileOrphanedWorkspaceWorktrees()).toBe(1); expect(await settled.reconcileOrphanedWorkspaceWorktrees()).toBe(0); diff --git a/packages/engine/src/__tests__/workspace-merger-lease.test.ts b/packages/engine/src/__tests__/workspace-merger-lease.test.ts index 4ffb4d611d..6fd3be4d1e 100644 --- a/packages/engine/src/__tests__/workspace-merger-lease.test.ts +++ b/packages/engine/src/__tests__/workspace-merger-lease.test.ts @@ -58,6 +58,18 @@ function createStore(task: Task): TaskStore & RecordingStore { Object.assign(store.task, patch); return undefined; }), + mergeWorkspaceWorktreeEntry: vi.fn(async ( + _id: string, + repoRelPath: string, + patch: Partial[string]>, + options?: { requireExistingEntry?: boolean }, + ) => { + const current = store.task.workspaceWorktrees ?? {}; + const existing = current[repoRelPath]; + if (options?.requireExistingEntry && !existing) return store.task; + store.task.workspaceWorktrees = { ...current, [repoRelPath]: { ...existing, ...patch } }; + return store.task; + }), logEntry: vi.fn().mockResolvedValue(undefined), appendAgentLog: vi.fn().mockResolvedValue(undefined), getTask: vi.fn(async () => store.task), diff --git a/packages/engine/src/__tests__/workspace-merger.test.ts b/packages/engine/src/__tests__/workspace-merger.test.ts index c9c16e5e02..60d4f5c38b 100644 --- a/packages/engine/src/__tests__/workspace-merger.test.ts +++ b/packages/engine/src/__tests__/workspace-merger.test.ts @@ -58,6 +58,7 @@ function createStore(settings: Record = {}): TaskStore & Record emitted, getSettings: vi.fn().mockResolvedValue({ autoMerge: false, ...settings }), updateTask: vi.fn().mockResolvedValue(undefined), + mergeWorkspaceWorktreeEntry: vi.fn().mockResolvedValue(undefined), logEntry: vi.fn().mockResolvedValue(undefined), appendAgentLog: vi.fn().mockResolvedValue(undefined), // FNXC:Test 2026-06-24-23:50: mergeAndReview reads store.getTask().comments for merge/review @@ -286,11 +287,12 @@ describeIfGit("landWorkspaceTask — per-repo merge loop (Phase C U1)", () => { expect(byRepo["repo-a"].status).toBe("landed"); expect(byRepo["repo-b"].status).toBe("failed"); expect(byRepo["repo-b"].error).toMatch(/conflict/i); - expect(store.updateTask).toHaveBeenCalledWith(TASK_ID, expect.objectContaining({ - workspaceWorktrees: expect.objectContaining({ - "repo-b": expect.objectContaining({ landFailure: expect.objectContaining({ message: expect.stringMatching(/conflict/i), branch: BRANCH }) }), - }), - })); + expect(store.mergeWorkspaceWorktreeEntry).toHaveBeenCalledWith( + TASK_ID, + "repo-b", + expect.objectContaining({ landFailure: expect.objectContaining({ message: expect.stringMatching(/conflict/i), branch: BRANCH }) }), + { requireExistingEntry: true }, + ); // Repo A landed locally (its ref advanced). expect(fx.git("repo-a", "git rev-parse refs/heads/main")).not.toBe(tipABefore); diff --git a/packages/engine/src/__tests__/worktree-acquisition-workspace.test.ts b/packages/engine/src/__tests__/worktree-acquisition-workspace.test.ts index f9557ad63d..e52898908f 100644 --- a/packages/engine/src/__tests__/worktree-acquisition-workspace.test.ts +++ b/packages/engine/src/__tests__/worktree-acquisition-workspace.test.ts @@ -34,17 +34,43 @@ function git(repo: string, command: string): string { */ function makeFakeStore( task: Task, - options: { failWhen?: (patch: Partial) => boolean } = {}, + options: { + failWhen?: (patch: Partial) => boolean; + beforeWorkspaceMerge?: (repoRelPath: string) => Promise; + } = {}, ): { store: TaskStore; current: () => Task; logs: string[]; patches: Partial[] } { let current = task; const logs: string[] = []; const patches: Partial[] = []; const store = { async updateTask(id: string, patch: Partial): Promise { + // Deliberately retain wholesale replacement: the concurrent regression below must fail + // if production returns to updateTask({ workspaceWorktrees }) instead of the key merge. patches.push(patch); if (options.failWhen?.(patch)) throw new Error("injected update failure"); if (id === current.id) current = { ...current, ...patch }; }, + async mergeWorkspaceWorktreeEntry( + id: string, + repoRelPath: string, + patch: Partial[string]>, + mergeOptions?: { requireExistingEntry?: boolean; clearSingularWorktree?: boolean }, + ): Promise { + if (id !== current.id) throw new Error(`Task ${id} not found`); + await options.beforeWorkspaceMerge?.(repoRelPath); + // Read only after the deterministic gate: this mirrors the store primitive's locked fresh read. + const workspaceWorktrees = current.workspaceWorktrees ?? {}; + const existing = workspaceWorktrees[repoRelPath]; + if (mergeOptions?.requireExistingEntry && !existing) return current; + const mergedPatch: Partial = { + workspaceWorktrees: { ...workspaceWorktrees, [repoRelPath]: { ...existing, ...patch } }, + ...(mergeOptions?.clearSingularWorktree ? { worktree: undefined, branch: undefined } : {}), + }; + patches.push(mergedPatch); + if (options.failWhen?.(mergedPatch)) throw new Error("injected update failure"); + current = { ...current, ...mergedPatch }; + return current; + }, async logEntry(_id: string, message: string): Promise { logs.push(message); }, @@ -376,6 +402,60 @@ describeIfGit("acquireWorkspaceRepoWorktree (U2 per-repo hardening)", { timeout: expect(patches.every((patch) => !patch.worktree && !patch.branch)).toBe(true); }); + it("retains both sibling entries when different repo acquisitions overlap deterministically", async () => { + fixture = await createWorkspaceFixture(["repo-a", "repo-b"]); + let firstAtMerge!: () => void; + const firstReachedMerge = new Promise((resolve) => { firstAtMerge = resolve; }); + let releaseFirst!: () => void; + const release = new Promise((resolve) => { releaseFirst = resolve; }); + let mergeCalls = 0; + const { store, current } = makeFakeStore(makeTask("FN-7-concurrent"), { + beforeWorkspaceMerge: async () => { + mergeCalls += 1; + if (mergeCalls === 1) { + firstAtMerge(); + await release; + } else { + // The second acquisition reaches the atomic merge before the first writes. + releaseFirst(); + } + }, + }); + const registry = new ActiveSessionRegistry(); + + const first = acquireWorkspaceRepoWorktree({ repoRelPath: "repo-a", workspaceRootDir: fixture.rootDir, task: current(), store, settings: SETTINGS, registry }); + await firstReachedMerge; + const second = acquireWorkspaceRepoWorktree({ repoRelPath: "repo-b", workspaceRootDir: fixture.rootDir, task: current(), store, settings: SETTINGS, registry }); + const [a, b] = await Promise.all([first, second]); + + expect(a.alreadyAcquired).toBe(false); + expect(b.alreadyAcquired).toBe(false); + expect(current().workspaceWorktrees).toMatchObject({ + "repo-a": { worktreePath: a.worktreePath, branch: a.branch }, + "repo-b": { worktreePath: b.worktreePath, branch: b.branch }, + }); + expect(current().worktree).toBeFalsy(); + expect(current().branch).toBeFalsy(); + }); + + it("keeps same-repo concurrent acquisition exclusive while the winning merge is pending", async () => { + fixture = await createWorkspaceFixture(["repo-a"]); + let reachedMerge!: () => void; + const firstReachedMerge = new Promise((resolve) => { reachedMerge = resolve; }); + let releaseFirst!: () => void; + const release = new Promise((resolve) => { releaseFirst = resolve; }); + const { store, current } = makeFakeStore(makeTask("FN-7-same-repo"), { + beforeWorkspaceMerge: async () => { reachedMerge(); await release; }, + }); + const registry = new ActiveSessionRegistry(); + const first = acquireWorkspaceRepoWorktree({ repoRelPath: "repo-a", workspaceRootDir: fixture.rootDir, task: current(), store, settings: SETTINGS, registry }); + await firstReachedMerge; + await expect(acquireWorkspaceRepoWorktree({ repoRelPath: "repo-a", workspaceRootDir: fixture.rootDir, task: makeTask("FN-7-same-repo-loser"), store, settings: SETTINGS, registry })).rejects.toBeInstanceOf(WorkspaceRepoAcquireBusyError); + releaseFirst(); + const winner = await first; + expect(current().workspaceWorktrees?.["repo-a"]?.worktreePath).toBe(winner.worktreePath); + }); + it("re-acquires a dead remembered workspace entry without singular persistence", async () => { fixture = await createWorkspaceFixture(["repo-a"]); const initial = { diff --git a/packages/engine/src/merge/merger-ai.ts b/packages/engine/src/merge/merger-ai.ts index 80fbc34c63..c4fcdac188 100644 --- a/packages/engine/src/merge/merger-ai.ts +++ b/packages/engine/src/merge/merger-ai.ts @@ -2237,9 +2237,8 @@ export { isRepoLanded }; /** * FNXC:Workspace 2026-06-22-00:30 (Phase C U2, KTD3): - * Persist one sub-repo's `landedSha` with a FRESH-read-then-merge so a concurrent - * sibling-entry write is not clobbered (Phase A/B per-repo `workspaceWorktrees` - * pattern). Re-read the latest task, merge only this repo's entry, write the whole map. + * Persist one sub-repo's `landedSha` through the store's advisory-locked per-key merge, + * so a concurrent sibling-entry acquisition or landing cannot be clobbered. * * FNXC:Workspace 2026-06-22-04:10 (Phase C review A1 — do NOT swallow the DB write): * Previously the `store.updateTask(...)` was `.catch(() => undefined)`. That swallow is the @@ -2251,6 +2250,10 @@ export { isRepoLanded }; * trailer ancestor-fallback (A1) recognises the actually-landed repo and skips it (no double * squash). We DELIBERATELY do not swallow the `getTask` read either-way: a failed read leaves * `landedSha` unrecorded for the same reason, so it must also escalate. + * + * FNXC:Workspace 2026-08-15-07:51: requireExistingEntry preserves the vanished-entry no-op, + * while mergeWorkspaceWorktreeEntry owns the cross-process atomic map merge. Do not substitute + * updateTask with a reconstructed workspaceWorktrees map here. */ async function persistRepoLandedSha( store: TaskStore, @@ -2258,14 +2261,14 @@ async function persistRepoLandedSha( repoRel: string, landedSha: string, ): Promise { - const latest = await store.getTask(taskId); - const current = latest?.workspaceWorktrees ?? {}; - const entry = current[repoRel]; - if (!entry) return; // entry vanished — nothing to merge into // FNXC:Workspace 2026-08-15-06:45: a new landing is strictly after its revert boundary, // so clear that invalidation marker while retaining the fresh landedSha as normal proof. - const next = { ...current, [repoRel]: { ...entry, landedSha, landFailure: undefined, revertBoundarySha: undefined } }; - await store.updateTask(taskId, { workspaceWorktrees: next }); + await store.mergeWorkspaceWorktreeEntry( + taskId, + repoRel, + { landedSha, landFailure: undefined, revertBoundarySha: undefined }, + { requireExistingEntry: true }, + ); } /** diff --git a/packages/engine/src/merge/workspace-land-failure.ts b/packages/engine/src/merge/workspace-land-failure.ts index a011815618..c9f0618769 100644 --- a/packages/engine/src/merge/workspace-land-failure.ts +++ b/packages/engine/src/merge/workspace-land-failure.ts @@ -3,7 +3,11 @@ import type { TaskStore } from "@fusion/core"; /** * FNXC:Workspace 2026-08-15-07:05: * Persist a display-only per-repository landing failure without altering merge control flow. - * A fresh read preserves concurrent updates to sibling workspace entries and callers swallow errors. + * + * FNXC:Workspace 2026-08-15-08:00: + * This is a per-key mutation, so it must use the advisory-locked store merge rather than + * reconstructing workspaceWorktrees from a stale read. `requireExistingEntry` retains the + * absent-entry no-op while concurrent acquisition and landing keep every sibling entry. */ export async function persistWorkspaceRepoLandFailure( store: TaskStore, @@ -11,11 +15,10 @@ export async function persistWorkspaceRepoLandFailure( repoRel: string, failure: { message: string; at: string; branch?: string }, ): Promise { - const latest = await store.getTask(taskId); - const current = latest?.workspaceWorktrees ?? {}; - const entry = current[repoRel]; - if (!entry) return; - await store.updateTask(taskId, { - workspaceWorktrees: { ...current, [repoRel]: { ...entry, landFailure: failure } }, - }); + await store.mergeWorkspaceWorktreeEntry( + taskId, + repoRel, + { landFailure: failure }, + { requireExistingEntry: true }, + ); } diff --git a/packages/engine/src/self-healing.ts b/packages/engine/src/self-healing.ts index ec305dc495..7586a6749b 100644 --- a/packages/engine/src/self-healing.ts +++ b/packages/engine/src/self-healing.ts @@ -10661,9 +10661,18 @@ const movedTask = await this.store.moveTask(task.id, completeLane); landed evidence for operator recovery and later safe deletion; deleting the whole entry would turn a safe retain into a permanent leak. */ - const worktrees = { ...(task.workspaceWorktrees ?? {}) }; - if (worktrees[repoRel]) worktrees[repoRel] = { ...worktrees[repoRel], worktreePath: "" }; - await this.store.updateTask(task.id, { workspaceWorktrees: worktrees }); + /* + FNXC:Workspace 2026-08-15-08:00: + Teardown settles one durable repository entry. The store-level advisory-locked merge + refreshes the map under the task's composite project scope, so this best-effort sweep + cannot erase a sibling acquisition or landed-SHA mutation that raced its stale task scan. + */ + await this.store.mergeWorkspaceWorktreeEntry( + task.id, + repoRel, + { worktreePath: "" }, + { requireExistingEntry: true }, + ); } catch { /* soft-deleted rows may reject best-effort settlement */ } } try { await createRunAuditor(this.store, { runId: generateSyntheticRunId("self-healing-orphaned-workspace-worktree", task.id), agentId: "self-healing", taskId: task.id, taskLineageId: task.lineageId, phase: "reconcile-orphaned-workspace-worktree" }).database({ type: "task:reconcile-orphaned-workspace-worktree", target: task.id, metadata: { taskId: task.id, repo: repoRel, worktreePath, success: settled, reason: failed ? "git-teardown-failed" : "settled", lane, worktreeOutcome: worktreeGone ? "gone" : "present", pruned, branch: entry.branch, branchOutcome, attempt } }); } catch { /* audit best-effort */ } diff --git a/packages/engine/src/worktree/worktree-acquisition.ts b/packages/engine/src/worktree/worktree-acquisition.ts index 23f880ed51..1ed88e5665 100644 --- a/packages/engine/src/worktree/worktree-acquisition.ts +++ b/packages/engine/src/worktree/worktree-acquisition.ts @@ -1518,20 +1518,12 @@ export async function acquireWorkspaceRepoWorktree( } /* - FNXC:Workspace 2026-06-21-22:30: - F5 — re-read the task fresh immediately before building the merged - workspaceWorktrees map. store.updateTask wholesale-replaces the map, and the - `task` snapshot was read earlier; two sequential acquires for DIFFERENT sub-repos - in one task would otherwise clobber a sibling's entry. Merging into the LATEST map - closes the common sequential-tool-call case. NOTE: a fully-atomic store-level - per-repo merge is the complete fix (it also covers truly-concurrent writes); it is - deferred to Phase B, which exercises multi-repo acquisition. + FNXC:Workspace 2026-08-15-07:51: + F5 Phase B is implemented by mergeWorkspaceWorktreeEntry: its advisory-locked, + per-key database merge retains sibling sub-repo entries across concurrent processes. + Do not restore a wholesale workspaceWorktrees update here; that reopens the silent + sibling-clobber race which leaves an on-disk worktree invisible to workspace landing. */ - const latest = await store.getTask(task.id); - const updated: Record = { - ...(latest.workspaceWorktrees ?? {}), - [repoRelPath]: { worktreePath: result.worktreePath, branch: result.branch, baseCommitSha }, - }; /* FNXC:Workspace 2026-08-15-04:28: F10 — this is the one durable acquisition-state write. The helper suppresses every earlier @@ -1540,7 +1532,12 @@ export async function acquireWorkspaceRepoWorktree( never makes dashboard workspace rendering, self-healing, or executor dispatch read it as single-repo. */ - await store.updateTask(task.id, { workspaceWorktrees: updated, worktree: null, branch: null }); + await store.mergeWorkspaceWorktreeEntry( + task.id, + repoRelPath, + { worktreePath: result.worktreePath, branch: result.branch, baseCommitSha }, + { clearSingularWorktree: true }, + ); return { worktreePath: result.worktreePath, branch: result.branch, baseCommitSha, alreadyAcquired: false }; } catch (err) {