fix(review): Phase C merge-loop hardening — double-land, lease clobber, retry storm
5-persona review of the Phase-C per-repo merge loop. No P0; the no-push invariant and retry/park accounting verified clean. Fixed: Land mechanics (merger-ai.ts / active-session-registry.ts): - persistRepoLandedSha no longer swallows the DB write: a failed landedSha write after the ref advanced now escalates to WorkspacePartialLandError so the engine parks/retries instead of silently re-landing (duplicate squash). isRepoLanded gains a landedSha-independent fallback — it scans the integration ref for this task's Fusion-Task-Id trailer (a squash commit is NOT a branch descendant, so a branch-ancestor check is provably wrong), so an actually-landed repo is skipped on retry. - The land lease is now taskId-aware across kinds: any foreign-task holder on a sub-repo path is contention (a merging task can't run over an executing task's acquire lease), and registerPath throws ActiveSessionPathHeldByForeignTaskError instead of silently clobbering a different task's entry. - The per-repo loop is wrapped in try/finally(setStatus(null)) so the busy/partial throws can't leave the task stuck 'merging'. WorkspacePartialLandError is a real exported class (not a .name-mutated Error). finalizeWorkspaceTask re-reads fresh and no longer swallows the mergeDetails write (TOCTOU). isRepoLanded exported for Phase D. Dispatch + doors (project-engine.ts / dashboard.ts / task.ts / @fusion/core): - getTask-null in the partial-land catch fails closed (park) instead of defaulting retries to 0 and scheduling an indefinite retry storm. - The merge-confirmed reachability fast-path skips workspace tasks (its representative commitSha is a sub-repo squash sha, unreachable in the root cwd — it was demoting fully-merged tasks); they're verified by per-repo landedSha. - The CLI/dashboard merge doors now return merged:true on full land (were hardcoded merged:false). WorkspaceRepoLandBusyError re-enqueues with backoff WITHOUT burning the mergeRetries quota (bounded busy counter) so contention can't park a healthy task. Backoff capped at 60s. shouldRetryWorkspacePartialLand folded into shouldRetryAutoMergeConflict. Catch switched to instanceof. New canonical isWorkspaceTask predicate in @fusion/core. Gate green: build, typecheck, lint, test:gate (649+58); workspace-merger + oracle + project-engine 174. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsxdsm
@@ -0,0 +1,7 @@
|
||||
---
|
||||
"@runfusion/fusion": patch
|
||||
---
|
||||
|
||||
Harden the workspace per-repo land loop against partial-failure races. A lost `landedSha` DB write after a sub-repo's integration ref already advanced no longer silently continues — it escalates to a retryable partial-land error, and the landed predicate now recognizes an already-landed repo via its `Fusion-Task-Id` trailer on retry, so a re-run never produces a second squash commit. The land lease is now taskId-aware across registry kinds: a merging task can no longer clobber an executing task's acquire lease on a shared sub-repo (any foreign-task holder is treated as contention), and the active-session registry rejects foreign-task overwrites instead of silently clobbering. The transient `merging` status is always reset before any throw escapes the land loop (no stuck-`merging` leak), and finalize re-reads the latest task and no longer swallows the merge-details persist failure (no finalizing on a stale row).
|
||||
|
||||
Harden the workspace merge dispatch and user-facing merge doors. The partial-land retry catch now fails closed when the task row can't be read (DB outage no longer triggers an indefinite retry storm). The merge-confirmed reachability fast-path skips workspace tasks (whose recorded commitSha lives in a sub-repo, not the workspace root) so a fully-landed workspace task is no longer demoted/parked. The dashboard and CLI merge doors now report `merged: true` (and `mergeConfirmed`/`commitSha`) when a workspace fully lands, mirroring the engine result. Transient sub-repo land-lease contention (`WorkspaceRepoLandBusyError`) is re-enqueued with capped backoff on a separate bounded counter instead of burning the merge-retry quota, so pure contention can't park a never-failed task. Retry backoff is capped at 60s.
|
||||
@@ -1319,12 +1319,19 @@ export async function runDashboard(port: number, opts: { paused?: boolean; dev?:
|
||||
agentStore,
|
||||
});
|
||||
const latest = await store.getTask(taskId).catch(() => mergeTask!);
|
||||
// U1 does not finalize the workspace task (finalize-once move-to-done is U2);
|
||||
// report merged=false until then.
|
||||
// FNXC:Workspace 2026-06-22-05:10 (Phase C review B3):
|
||||
// landWorkspaceTask now finalizes the workspace task to done on allLanded (Phase C U2),
|
||||
// so the merge door must report merged=true when the workspace fully landed — mirroring
|
||||
// the engine dispatch's MergeResult. The first landed sub-repo's landedSha is the recorded
|
||||
// commitSha (same convention finalizeWorkspaceTask uses). On a partial land, merged stays
|
||||
// false and the partial-land error surfaces on the task log.
|
||||
const landedSha = workspaceResult.repos.find((r) => r.status === "landed")?.landedSha;
|
||||
return {
|
||||
task: latest ?? mergeTask!,
|
||||
branch: getTaskBranchName(taskId),
|
||||
merged: false,
|
||||
merged: workspaceResult.allLanded,
|
||||
mergeConfirmed: workspaceResult.allLanded || undefined,
|
||||
commitSha: workspaceResult.allLanded ? landedSha : undefined,
|
||||
worktreeRemoved: false,
|
||||
branchDeleted: false,
|
||||
error: workspaceResult.allLanded ? undefined : "partial workspace land — see task log",
|
||||
|
||||
@@ -872,8 +872,13 @@ export async function runTaskMerge(id: string, projectName?: string) {
|
||||
: `failed: ${repo.error ?? "unknown"}`;
|
||||
console.log(` ${repo.status === "failed" ? "✗" : "✓"} ${repo.repo}: ${label}`);
|
||||
}
|
||||
// U1 does not move the workspace task to done (finalize-once is U2).
|
||||
console.log(`\n ${workspaceResult.allLanded ? "✓ All sub-repos landed" : "✗ Partial land — see failures above"} (task remains in review until U2)\n`);
|
||||
// FNXC:Workspace 2026-06-22-05:10 (Phase C review B3):
|
||||
// landWorkspaceTask now finalizes the workspace task to done on allLanded (Phase C U2),
|
||||
// so report it as merged rather than "remains in review until U2". A partial land leaves
|
||||
// the task in review (landed repos stay landed locally) and exits non-zero.
|
||||
console.log(
|
||||
`\n ${workspaceResult.allLanded ? "✓ All sub-repos landed — task finalized to done" : "✗ Partial land — see failures above (task remains in review; landed repos stay landed locally)"}\n`,
|
||||
);
|
||||
if (!workspaceResult.allLanded) process.exit(1);
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
export { COLUMNS, DEFAULT_COLUMN, isColumn, normalizeColumn, COLUMN_LABELS, COLUMN_DESCRIPTIONS, VALID_TRANSITIONS, DEFAULT_SETTINGS, DEFAULT_GLOBAL_SETTINGS, DEFAULT_PROJECT_SETTINGS, GLOBAL_SETTINGS_KEYS, PROJECT_SETTINGS_KEYS, isGlobalSettingsKey, isProjectSettingsKey, isMergeRequestContractShadowEnabled, resolvePersistAgentThinkingLog, THINKING_LEVELS, THEME_MODES, COLOR_THEMES, SUPPORTED_LOCALES, DEFAULT_LOCALE, isLocale, WORKFLOW_STEP_TEMPLATES, AGENT_PERMISSIONS, PERMANENT_AGENT_ACTION_CATEGORIES, AGENT_PERMISSION_POLICY_ACTION_CATEGORIES, AGENT_PROVISIONING_APPROVAL_MODES, SANDBOX_PROVISIONING_APPROVAL_MODES, AGENT_PERMISSION_POLICY_PRESET_IDS, LEGACY_AGENT_PERMISSION_POLICY_ACTION_CATEGORY_ALIASES, APPROVAL_REQUEST_STATUSES, APPROVAL_REQUEST_AUDIT_EVENT_TYPES, normalizeApprovalRequestActionCategory, isValidApprovalRequestTransition, agentToConfigSnapshot, diffConfigSnapshots, isEphemeralAgent, hasAgentIdentity, CheckoutConflictError, DEFAULT_HEARTBEAT_PROCEDURE_PATH, getDefaultHeartbeatProcedurePath, EXECUTION_MODES, DEFAULT_EXECUTION_MODE, TASK_PRIORITIES, DEFAULT_TASK_PRIORITY, WORKFLOW_WORK_ITEM_KINDS, WORKFLOW_WORK_ITEM_STATES, HIGH_FANOUT_BLOCKER_TODO_THRESHOLD, STALE_HIGH_FANOUT_BLOCKER_AGE_THRESHOLD_MS, DASHBOARD_USER_ID, normalizeMessageParticipant, validateMessageMetadata, validateDockerNodeConfig, sanitizeDockerNodeConfigForResponse, normalizeMergeIntegrationWorktreeMode, normalizeMergeAdvanceAutoSyncMode, MERGE_ADVANCE_AUTO_SYNC_MODES, normalizeMergeConflictStrategy, normalizeMergeStrategyOverlapBehavior, normalizePostMergeAuditMode, POST_MERGE_AUDIT_MODES, normalizeMergeAuditAutoRecovery, MERGE_AUDIT_AUTO_RECOVERY_MODES, normalizeMergerMode, MERGER_MODES, normalizeAutoRecovery, AUTO_RECOVERY_MODES, buildResearchDocumentKey, REPO_OVERRIDE_RE, SHARED_STATE_SNAPSHOT_VERSION, sanitizeCliAgentSettings, sanitizeCliAgentsSettings, CLI_AGENT_ADAPTER_IDS, CLI_AGENT_AUTONOMY_MODES } from "./types.js";
|
||||
export type { Column, ColumnId, IssueInfo, IssueState, TaskSourceIssue, PrInfo, PrConflictState, PrConflictDiagnostics, PrCheckState, PrCheckStatus, PrStatus, BranchGroup, BranchGroupCreateInput, BranchGroupUpdate, BranchGroupPrState, Task, TaskTokenUsage, TaskTokenUsagePerModel, TaskAttachment, TaskComment, TaskCommentInput, TaskDocument, TaskDocumentRevision, TaskDocumentCreateInput, TaskDocumentWithTask, ArtifactType, Artifact, ArtifactCreateInput, ArtifactWithTask, TaskCreateInput, MeshReplicatedTaskCreatePayload, MeshReplicatedTaskApplyResult, TaskSource, SourceType, TaskDetail, RetrySummary, InboxTask, TodoList, TodoItem, TodoListCreateInput, TodoListUpdateInput, TodoItemCreateInput, TodoItemUpdateInput, TodoListWithItems, AgentLogEntry, AgentLogType, AgentRole, BoardConfig, DistributedTaskIdReserveInput, DistributedTaskIdReserveResult, DistributedTaskIdCommitInput, DistributedTaskIdCommitResult, DistributedTaskIdAbortInput, DistributedTaskIdAbortResult, DistributedTaskIdStateInput, DistributedTaskIdStateResult, AutostashOrphanRecord, AutostashOutcome, MergeDetails, MergeResult, MergeIntegrationWorktreeMode, MergeAdvanceAutoSyncMode, MergeConflictStrategy, CanonicalMergeConflictStrategy, MergeStrategyOverlapBehavior, PostMergeAuditMode, MergeAuditAutoRecoveryMode, MergerMode, MergerSettings, AutoRecoveryMode, AutoRecoveryFailureClass, AutoRecoverySettings, DirectMergeCommitStrategy, Settings, GlobalSettings, ProjectSettings, SecretsEnvConfig, WebSearchBackend, ResearchEnabledSources, ResearchGlobalDefaults, ResearchProjectLimits, ResearchProjectSettings, SandboxBackendName, SandboxFailureMode, SandboxPolicy, SandboxProjectSettings, EvalFollowUpPolicy, EvalProjectSettings, ResolvedEvalSettings, SettingsScope, DaemonTokenSettings, TaskStep, StepStatus, TaskLogEntry, RunMutationContext, ActivityLogEntry, ActivityEventType, ThinkingLevel, ThemeMode, ColorTheme, Locale, ExecutionMode, TaskPriority, MergeQueueEntry, MergeQueueEnqueueOptions, MergeQueueAcquireOptions, MergeQueueReleaseOutcome, MergeRequestState, MergeRequestRecord, MergeRequestWorkflowProjectionOptions, CompletionHandoffMarker, WorkflowWorkItem, WorkflowWorkItemDueFilter, WorkflowWorkItemKind, WorkflowWorkItemState, WorkflowWorkItemTransitionPatch, WorkflowWorkItemUpsertInput, HandoffEvidence, HandoffToReviewOptions, UnavailableNodePolicy, OwningNodeHandoffPolicy, PlanningQuestion, PlanningSummary, PlanningResponse, PlanningQuestionType, ArchivedTaskEntry, BatchStatusRequest, BatchStatusResponse, BatchStatusEntry, BatchStatusResult, GithubIssueAction, ModelPreset, WorkflowStep, WorkflowStepMode, WorkflowStepGateMode, WorkflowStepPhase, WorkflowStepInput, WorkflowStepResult, WorkflowStepTemplate, Agent, OrgTreeNode, AgentState, AgentDetail, AgentCreateInput, AgentUpdateInput, AgentApiKey, AgentApiKeyCreateResult, AgentCapability, AgentPromptTemplate, AgentPromptsConfig, AgentPermission, PermanentAgentActionCategory, PermanentAgentSensitiveActionCategory, PermanentAgentGatingContext, AgentPermissionPolicy, AgentPermissionPolicyRules, AgentPermissionPolicyActionCategory, AgentProvisioningApprovalMode, SandboxProvisioningApprovalMode, LegacyAgentPermissionPolicyActionCategory, ApprovalRequestActionCategoryInput, ApprovalRequestActionCategory, AgentPermissionPolicyDisposition, AgentPermissionPolicyPresetId, ApprovalRequestStatus, ApprovalRequestAuditEventType, ApprovalRequestActorSnapshot, ApprovalRequestTargetAction, ApprovalRequestAuditEvent, ApprovalRequest, ApprovalRequestCreateInput, ApprovalRequestDecisionInput, ApprovalRequestCompletionInput, ApprovalRequestListInput, TaskAssignSource, AgentAccessState, AgentHeartbeatConfig, AgentBudgetConfig, AgentBudgetStatus, InstructionsBundleConfig, MessageResponseMode, AgentHeartbeatEvent, AgentHeartbeatRun, BlockedStateSnapshot, HeartbeatInvocationSource, AgentTaskSession, AgentRating, AgentRatingSummary, AgentRatingInput, AgentConfigSnapshot, RevisionFieldDiff, AgentConfigRevision, AgentStats, ReflectionTrigger, ReflectionMetrics, AgentReflection, AgentPerformanceSummary, NtfyNotificationEvent, NotificationEvent, NotificationPayload, NotificationProviderConfig, CustomProvider, SteeringComment, ParticipantType, MessageType, Message, MessageCreateInput, MessageFilter, MessageMetadata, MessageReplyReference, Mailbox, CheckoutLease, CheckoutClaimPrecondition, TaskClaimRow, CentralClaimStore, RunAuditDomain, RunAuditEvent, RunAuditEventInput, RunAuditEventFilter, AgentMemoryInclusionMode, HeartbeatPromptTemplate, HeartbeatScopeDisciplineMode, WorktrunkSettings, WorktrunkOnFailure, TaskBranchContext, CliAgentSettings } from "./types.js";
|
||||
export { AGENT_VALID_TRANSITIONS, DUPLICATE_OF_METADATA_KEY, assertNotWorkspaceTaskMerge, WorkspaceTaskMergeError } from "./types.js";
|
||||
export { AGENT_VALID_TRANSITIONS, DUPLICATE_OF_METADATA_KEY, assertNotWorkspaceTaskMerge, isWorkspaceTask, WorkspaceTaskMergeError } from "./types.js";
|
||||
export {
|
||||
resolveEntryPointBranchAssignment,
|
||||
sanitizeBranchSegment,
|
||||
|
||||
@@ -2662,14 +2662,28 @@ export class WorkspaceTaskMergeError extends Error {
|
||||
* @param task the task about to enter a merge path
|
||||
*/
|
||||
export function assertNotWorkspaceTaskMerge(task: Pick<Task, "id" | "workspaceWorktrees">): void {
|
||||
const worktrees = task.workspaceWorktrees;
|
||||
if (worktrees && Object.keys(worktrees).length > 0) {
|
||||
if (isWorkspaceTask(task)) {
|
||||
throw new WorkspaceTaskMergeError(
|
||||
`Workspace task ${task.id} cannot merge until per-repo merge support (master-plan U6) lands`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/*
|
||||
FNXC:Workspace 2026-06-22-05:10 (Phase C review B5/B7-dep — canonical workspace predicate):
|
||||
A workspace-mode task is identified by having at least one `workspaceWorktrees` entry
|
||||
(one git worktree per sub-repo). This single predicate replaces the inlined
|
||||
`!!task.workspaceWorktrees && Object.keys(task.workspaceWorktrees).length > 0` that was
|
||||
copy-pasted across the engine merge dispatch and the merge-confirmed reachability fast-path
|
||||
(B2). It lives in @fusion/core so the engine, store, and CLI doors share ONE definition.
|
||||
The dashboard keeps its own local `isWorkspaceTask` (WorkspaceWorktreesSummary, UI-only) —
|
||||
this core export is for engine/CLI use.
|
||||
*/
|
||||
export function isWorkspaceTask(task: Pick<Task, "workspaceWorktrees">): boolean {
|
||||
const worktrees = task.workspaceWorktrees;
|
||||
return !!worktrees && Object.keys(worktrees).length > 0;
|
||||
}
|
||||
|
||||
export type RetrySummary = {
|
||||
stuckKill: number;
|
||||
recovery: number;
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { beforeEach, describe, expect, it } from "vitest";
|
||||
import {
|
||||
activeSessionRegistry,
|
||||
reconcileSelfOwnedActiveSessionForRemoval,
|
||||
ActiveSessionPathHeldByForeignTaskError,
|
||||
} from "../active-session-registry.js";
|
||||
|
||||
describe("activeSessionRegistry", () => {
|
||||
@@ -28,15 +29,27 @@ describe("activeSessionRegistry", () => {
|
||||
expect(activeSessionRegistry.lookupByPath("/tmp/missing")).toBeNull();
|
||||
});
|
||||
|
||||
it("overwrites duplicate registration with warning", () => {
|
||||
const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {});
|
||||
// FNXC:Workspace 2026-06-22-04:10 (Phase C review A2 — taskId-aware lease across kinds):
|
||||
// registerPath must NOT silently clobber an entry held by a DIFFERENT task (that was the
|
||||
// cross-phase clobber bug: a merging task's land lease overwriting an executing task's
|
||||
// acquire lease on a shared sub-repo). A foreign-task overwrite now THROWS; the existing
|
||||
// foreign holder is preserved.
|
||||
it("rejects a foreign-task overwrite (does not clobber the held entry)", () => {
|
||||
activeSessionRegistry.registerPath("/tmp/w1", { taskId: "FN-1", kind: "executor", ownerKey: "FN-1" });
|
||||
activeSessionRegistry.registerPath("/tmp/w1", { taskId: "FN-2", kind: "workflow-step", ownerKey: "FN-2#workflow-step" });
|
||||
expect(() =>
|
||||
activeSessionRegistry.registerPath("/tmp/w1", { taskId: "FN-2", kind: "workflow-step", ownerKey: "FN-2#workflow-step" }),
|
||||
).toThrow(ActiveSessionPathHeldByForeignTaskError);
|
||||
// The original holder is untouched.
|
||||
expect(activeSessionRegistry.lookupByPath("/tmp/w1")?.taskId).toBe("FN-1");
|
||||
});
|
||||
|
||||
expect(activeSessionRegistry.lookupByPath("/tmp/w1")?.taskId).toBe("FN-2");
|
||||
expect(warnSpy).toHaveBeenCalledOnce();
|
||||
|
||||
warnSpy.mockRestore();
|
||||
// Same-task re-registration stays idempotent (an executor re-claiming/refreshing its own path).
|
||||
it("allows same-task re-registration (idempotent re-claim)", () => {
|
||||
activeSessionRegistry.registerPath("/tmp/w1", { taskId: "FN-1", kind: "executor", ownerKey: "FN-1" });
|
||||
expect(() =>
|
||||
activeSessionRegistry.registerPath("/tmp/w1", { taskId: "FN-1", kind: "step-session", ownerKey: "FN-1#step-session" }),
|
||||
).not.toThrow();
|
||||
expect(activeSessionRegistry.lookupByPath("/tmp/w1")?.kind).toBe("step-session");
|
||||
});
|
||||
|
||||
it("reconcileStaleSelfOwned returns no-entry when path is unregistered", () => {
|
||||
|
||||
@@ -1,6 +1,9 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import type { Task } from "@fusion/core";
|
||||
import { ProjectEngine, __resetDeterministicMergerModeDeprecationWarned } from "../project-engine.js";
|
||||
// Resolves to the vi.mock factory above (the mocked merger-ai exports the real-shaped
|
||||
// workspace land error classes so the dispatch's `instanceof` matching is exercised).
|
||||
import { WorkspacePartialLandError, WorkspaceRepoLandBusyError } from "../merger-ai.js";
|
||||
import { runtimeLog } from "../logger.js";
|
||||
import { TunnelProcessManager } from "../remote-access/tunnel-process-manager.js";
|
||||
import { NtfyNotifier } from "../notifier.js";
|
||||
@@ -19,6 +22,7 @@ const mocks = vi.hoisted(() => ({
|
||||
runtimeStop: vi.fn(async () => undefined),
|
||||
runtimeResumeAfterUnpause: vi.fn(async () => undefined),
|
||||
runAiMerge: vi.fn(),
|
||||
landWorkspaceTask: vi.fn(),
|
||||
execFile: vi.fn(),
|
||||
currentStore: null as Record<string, unknown> | null,
|
||||
notifierStart: vi.fn(async () => undefined),
|
||||
@@ -69,9 +73,42 @@ vi.mock("../merger.js", () => ({
|
||||
VerificationError: class VerificationError extends Error {},
|
||||
}));
|
||||
|
||||
vi.mock("../merger-ai.js", () => ({
|
||||
runAiMerge: mocks.runAiMerge,
|
||||
}));
|
||||
// FNXC:Workspace 2026-06-22-05:10 (Phase C review B7): the dispatch now matches the
|
||||
// workspace land errors via `instanceof`, and routes workspace tasks through
|
||||
// `landWorkspaceTask`. The mock must export REAL error classes (so `instanceof` is callable)
|
||||
// and a mockable `landWorkspaceTask`; otherwise `err instanceof WorkspacePartialLandError`
|
||||
// throws "not callable" and the workspace dispatch can't be exercised. The classes are
|
||||
// declared INSIDE the (hoisted) factory so they exist when the mock is evaluated.
|
||||
vi.mock("../merger-ai.js", () => {
|
||||
class WorkspaceRepoLandBusyError extends Error {
|
||||
public readonly retryable = true;
|
||||
constructor(
|
||||
public readonly repoRel: string,
|
||||
public readonly holderTaskId: string,
|
||||
public readonly requestingTaskId: string,
|
||||
) {
|
||||
super(`workspace sub-repo ${repoRel} land is in progress for task ${holderTaskId}`);
|
||||
this.name = "WorkspaceRepoLandBusyError";
|
||||
}
|
||||
}
|
||||
class WorkspacePartialLandError extends Error {
|
||||
public readonly retryable = true;
|
||||
constructor(
|
||||
public readonly landedCount: number,
|
||||
public readonly failedRepos: string[],
|
||||
message: string,
|
||||
) {
|
||||
super(message);
|
||||
this.name = "WorkspacePartialLandError";
|
||||
}
|
||||
}
|
||||
return {
|
||||
runAiMerge: mocks.runAiMerge,
|
||||
landWorkspaceTask: mocks.landWorkspaceTask,
|
||||
WorkspaceRepoLandBusyError,
|
||||
WorkspacePartialLandError,
|
||||
};
|
||||
});
|
||||
|
||||
vi.mock("node:child_process", async (importOriginal) => {
|
||||
const actual = await importOriginal<typeof import("node:child_process")>();
|
||||
@@ -1295,7 +1332,11 @@ describe("ProjectEngine U0 merge unification dispatch", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it("R7 guard: rejects a workspace-mode task at the engine merge entry point before any merge", async () => {
|
||||
// FNXC:Workspace 2026-06-22-05:10 (Phase C U1/U2 routing — supersedes the old R7 throw test):
|
||||
// A workspace-mode task no longer throws WorkspaceTaskMergeError at the engine dispatch; it
|
||||
// ROUTES to the per-repo land loop `landWorkspaceTask` (runAiMerge's R7 chokepoint stays as
|
||||
// defense-in-depth but is not the primary path). On a full land, the merge reports merged=true.
|
||||
it("routes a workspace-mode task to landWorkspaceTask (not runAiMerge) on full land", async () => {
|
||||
const mockStore = createMockStore({ ...baseSettings, autoMerge: true });
|
||||
mockStore.store.getTask.mockResolvedValue({
|
||||
id: "FN-WS",
|
||||
@@ -1303,58 +1344,217 @@ describe("ProjectEngine U0 merge unification dispatch", () => {
|
||||
paused: false,
|
||||
mergeRetries: 0,
|
||||
status: "queued",
|
||||
branch: "fusion/fn-ws",
|
||||
workspaceWorktrees: {
|
||||
"repo-a": { worktreePath: "/tmp/a", branch: "fusion/fn-ws-a" },
|
||||
"repo-b": { worktreePath: "/tmp/b", branch: "fusion/fn-ws-b" },
|
||||
},
|
||||
} as any);
|
||||
mocks.currentStore = mockStore.store;
|
||||
mocks.landWorkspaceTask.mockResolvedValue({
|
||||
allLanded: true,
|
||||
repos: [
|
||||
{ repo: "repo-a", status: "landed", landedSha: "aaaa1111", integrationBranch: "main" },
|
||||
{ repo: "repo-b", status: "landed", landedSha: "bbbb2222", integrationBranch: "main" },
|
||||
],
|
||||
} as any);
|
||||
|
||||
const engine = createEngine();
|
||||
await engine.start();
|
||||
await expect(engine.onMerge("FN-WS")).rejects.toThrow(
|
||||
/Workspace task FN-WS cannot merge until per-repo merge support \(master-plan U6\) lands/,
|
||||
);
|
||||
const result = await engine.onMerge("FN-WS");
|
||||
expect(mocks.landWorkspaceTask).toHaveBeenCalled();
|
||||
expect(mocks.runAiMerge).not.toHaveBeenCalled();
|
||||
expect(result.merged).toBe(true);
|
||||
await engine.stop();
|
||||
});
|
||||
});
|
||||
|
||||
/*
|
||||
FNXC:Workspace 2026-06-22-05:10 (Phase C review B1/B2/B4/B5):
|
||||
Merge DISPATCH hardening for workspace tasks. These drive the REAL ProjectEngine dispatch
|
||||
catch via the mocked merger-ai seam (landWorkspaceTask + the real-shaped error classes),
|
||||
asserting the failure modes the review flagged: fail-closed on getTask null (B1), the
|
||||
merge-confirmed reachability fast-path skipping workspace tasks (B2), busy-contention not
|
||||
burning the merge-retry quota (B4), and the capped backoff (B5). No real AI, no real git
|
||||
for the fast-path (the gate's git is asserted NOT to run for workspace tasks).
|
||||
*/
|
||||
describe("ProjectEngine workspace merge dispatch hardening (Phase C review)", () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
const workspaceTask = (overrides: Record<string, unknown> = {}) => ({
|
||||
id: "FN-WSH",
|
||||
column: "in-review",
|
||||
paused: false,
|
||||
mergeRetries: 0,
|
||||
status: "queued",
|
||||
branch: "fusion/fn-wsh",
|
||||
workspaceWorktrees: {
|
||||
"repo-a": { worktreePath: "/tmp/a", branch: "fusion/fn-wsh-a" },
|
||||
},
|
||||
...overrides,
|
||||
});
|
||||
|
||||
// B1: getTask returning null in the partial-land catch must FAIL CLOSED — no retry timer.
|
||||
it("B1: partial land with getTask null fails closed (parks failed, no retry timer)", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const mockStore = createMockStore({ ...baseSettings, autoMerge: true });
|
||||
// First getTask (dispatch routing) returns the workspace task; the catch's getTask
|
||||
// (after the throw) returns null to simulate a DB outage.
|
||||
mockStore.store.getTask
|
||||
.mockResolvedValueOnce(workspaceTask() as any) // dispatch routing read
|
||||
.mockResolvedValueOnce(workspaceTask() as any) // canMergeTask sweep read (if any)
|
||||
.mockResolvedValue(null as any); // catch-block read → DB outage
|
||||
mocks.currentStore = mockStore.store;
|
||||
mocks.landWorkspaceTask.mockRejectedValue(
|
||||
new WorkspacePartialLandError(0, ["repo-a"], "Workspace partial land for FN-WSH: 0 landed, 1 failed"),
|
||||
);
|
||||
|
||||
const engine = createEngine();
|
||||
await engine.start();
|
||||
const enqueueSpy = vi.spyOn(
|
||||
engine as unknown as { internalEnqueueMerge: (id: string) => void },
|
||||
"internalEnqueueMerge",
|
||||
);
|
||||
engine.enqueueMerge("FN-WSH");
|
||||
|
||||
// Drain microtasks until the catch parks the task (fail-closed path).
|
||||
await vi.waitFor(
|
||||
() => {
|
||||
expect(mockStore.store.updateTask).toHaveBeenCalledWith(
|
||||
"FN-WSH",
|
||||
expect.objectContaining({ status: "failed" }),
|
||||
);
|
||||
},
|
||||
{ timeout: 2000, interval: 5 },
|
||||
);
|
||||
|
||||
// No retry timer was scheduled, and no re-enqueue happened: advancing all timers
|
||||
// must not trigger another internalEnqueueMerge.
|
||||
enqueueSpy.mockClear();
|
||||
await vi.advanceTimersByTimeAsync(120_000);
|
||||
expect(enqueueSpy).not.toHaveBeenCalled();
|
||||
// It must NOT have incremented mergeRetries (it couldn't even read the row).
|
||||
expect(mockStore.store.updateTask).not.toHaveBeenCalledWith(
|
||||
"FN-WSH",
|
||||
expect.objectContaining({ mergeRetries: expect.anything(), status: null }),
|
||||
);
|
||||
|
||||
await engine.stop();
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
// B2: a merged workspace task (mergeConfirmed + sub-repo commitSha) must SKIP the root-cwd
|
||||
// reachability fast-path so it is finalized, not demoted/parked.
|
||||
it("B2: merge-confirmed workspace task skips the root-cwd reachability gate (not demoted)", async () => {
|
||||
const mockStore = createMockStore({ ...baseSettings, autoMerge: true });
|
||||
mockStore.store.getTask.mockResolvedValue(
|
||||
workspaceTask({
|
||||
status: null,
|
||||
mergeDetails: {
|
||||
mergeConfirmed: true,
|
||||
// A sub-repo squash sha — unreachable from the workspace ROOT cwd; the gate would
|
||||
// (wrongly) clear mergeConfirmed and demote the task if it ran here.
|
||||
commitSha: "deadbeefdeadbeefdeadbeefdeadbeefdeadbeef",
|
||||
mergeTargetBranch: "main",
|
||||
mergedAt: "2026-06-22T00:00:00.000Z",
|
||||
},
|
||||
}) as any,
|
||||
);
|
||||
mockStore.store.moveTask.mockResolvedValue(
|
||||
workspaceTask({ column: "done" }) as any,
|
||||
);
|
||||
mocks.currentStore = mockStore.store;
|
||||
// If the gate ran, it would invoke `git cat-file`. Make any git call fail so a gate
|
||||
// run would be observable (and would demote). We assert it is NOT called.
|
||||
mocks.execFile.mockImplementation((
|
||||
_file: string,
|
||||
_args: string[],
|
||||
optionsOrCb: unknown,
|
||||
callback?: (e: Error | null, r: { stdout: string; stderr: string }) => void,
|
||||
) => {
|
||||
const cb = (typeof optionsOrCb === "function" ? optionsOrCb : callback) as (
|
||||
e: Error | null,
|
||||
r: { stdout: string; stderr: string },
|
||||
) => void;
|
||||
cb(new Error("git should not be called for workspace fast-path"), { stdout: "", stderr: "" });
|
||||
return {} as never;
|
||||
});
|
||||
|
||||
const engine = createEngine();
|
||||
await engine.start();
|
||||
engine.enqueueMerge("FN-WSH");
|
||||
|
||||
await vi.waitFor(() => {
|
||||
expect(mockStore.store.emit).toHaveBeenCalledWith(
|
||||
"task:merged",
|
||||
expect.objectContaining({ merged: true }),
|
||||
);
|
||||
});
|
||||
|
||||
// The reachability gate's `git cat-file` must NOT have run (workspace skip).
|
||||
const gitCatFileCalls = (mocks.execFile.mock.calls as Array<[string, string[]]>).filter(
|
||||
(c) => Array.isArray(c[1]) && c[1][0] === "cat-file",
|
||||
);
|
||||
expect(gitCatFileCalls).toHaveLength(0);
|
||||
// The task must NOT have been demoted (mergeConfirmed cleared / status failed).
|
||||
expect(mockStore.store.updateTask).not.toHaveBeenCalledWith(
|
||||
"FN-WSH",
|
||||
expect.objectContaining({ status: "failed" }),
|
||||
);
|
||||
await engine.stop();
|
||||
});
|
||||
|
||||
// Regression: the auto-merge park for a WorkspaceTaskMergeError must set status:"failed",
|
||||
// not status:null. status:null + mergeRetries:0 passes every eligibility gate, so the
|
||||
// cooldown sweep re-enqueues the task every tick → tight re-throw/re-park loop. status:"failed"
|
||||
// makes canMergeTask short-circuit; manual retry still works (it bypasses canMergeTask).
|
||||
it("R7 auto-merge park: workspace task is parked status:'failed' so it is not re-enqueued", async () => {
|
||||
const mockStore = createMockStore({ ...baseSettings, autoMerge: true });
|
||||
mockStore.store.getTask.mockResolvedValue({
|
||||
id: "FN-WS-AUTO",
|
||||
column: "in-review",
|
||||
paused: false,
|
||||
mergeRetries: 0,
|
||||
status: "queued",
|
||||
workspaceWorktrees: {
|
||||
"repo-a": { worktreePath: "/tmp/a", branch: "fusion/fn-ws-a" },
|
||||
},
|
||||
} as any);
|
||||
mocks.currentStore = mockStore.store;
|
||||
|
||||
const engine = createEngine();
|
||||
await engine.start();
|
||||
// Auto-merge path (no manual resolver): the R7 door guard throws before runAiMerge,
|
||||
// and the dispatch catch parks the task.
|
||||
engine.enqueueMerge("FN-WS-AUTO");
|
||||
await vi.waitFor(() => {
|
||||
expect(mockStore.store.updateTask).toHaveBeenCalledWith(
|
||||
"FN-WS-AUTO",
|
||||
expect.objectContaining({ status: "failed", mergeRetries: 0 }),
|
||||
// B4 + B5: repeated WorkspaceRepoLandBusyError re-enqueues with capped backoff WITHOUT
|
||||
// consuming mergeRetries (pure contention does not park a never-failed task).
|
||||
it("B4/B5: busy contention re-enqueues with capped backoff, never burns mergeRetries", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const mockStore = createMockStore({ ...baseSettings, autoMerge: true });
|
||||
mockStore.store.getTask.mockResolvedValue(workspaceTask() as any);
|
||||
mocks.currentStore = mockStore.store;
|
||||
mocks.landWorkspaceTask.mockRejectedValue(
|
||||
new WorkspaceRepoLandBusyError("repo-a", "FN-OTHER", "FN-WSH"),
|
||||
);
|
||||
});
|
||||
expect(mocks.runAiMerge).not.toHaveBeenCalled();
|
||||
// Guard against regression to the re-enqueue loop (status:null park):
|
||||
expect(mockStore.store.updateTask).not.toHaveBeenCalledWith(
|
||||
"FN-WS-AUTO",
|
||||
expect.objectContaining({ status: null }),
|
||||
);
|
||||
await engine.stop();
|
||||
|
||||
const engine = createEngine();
|
||||
await engine.start();
|
||||
const enqueueSpy = vi.spyOn(
|
||||
engine as unknown as { internalEnqueueMerge: (id: string) => void },
|
||||
"internalEnqueueMerge",
|
||||
);
|
||||
engine.enqueueMerge("FN-WSH");
|
||||
|
||||
// The busy catch logs a WorkspaceRepoLandBusy entry then schedules a backoff timer.
|
||||
await vi.waitFor(
|
||||
() => {
|
||||
expect(mockStore.store.logEntry).toHaveBeenCalledWith(
|
||||
"FN-WSH",
|
||||
expect.stringContaining("busy"),
|
||||
"WorkspaceRepoLandBusy",
|
||||
);
|
||||
},
|
||||
{ timeout: 2000, interval: 5 },
|
||||
);
|
||||
|
||||
// It must NOT have written any mergeRetries increment (busy ≠ real failure).
|
||||
const burnedRetries = (mockStore.store.updateTask.mock.calls as Array<[string, Record<string, unknown>]>)
|
||||
.some((c) => c[0] === "FN-WSH" && typeof c[1]?.mergeRetries === "number");
|
||||
expect(burnedRetries).toBe(false);
|
||||
|
||||
// Drive several busy re-enqueues; the backoff must stay capped at 60s.
|
||||
enqueueSpy.mockClear();
|
||||
await vi.advanceTimersByTimeAsync(60_000); // first backoff (5s) fires → re-enqueue
|
||||
expect(enqueueSpy).toHaveBeenCalledWith("FN-WSH");
|
||||
|
||||
await engine.stop();
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -26,8 +26,19 @@ import { execSync } from "node:child_process";
|
||||
import { writeFileSync } from "node:fs";
|
||||
import path from "node:path";
|
||||
import type { Task, TaskStore } from "@fusion/core";
|
||||
import { landWorkspaceTask } from "../merger-ai.js";
|
||||
import { shouldRetryWorkspacePartialLand } from "../project-engine.js";
|
||||
import { landWorkspaceTask, WorkspacePartialLandError } from "../merger-ai.js";
|
||||
import { shouldRetryAutoMergeConflict } from "../project-engine.js";
|
||||
|
||||
/*
|
||||
FNXC:Workspace 2026-06-22-05:10 (Phase C review B6):
|
||||
`shouldRetryWorkspacePartialLand` was collapsed into `shouldRetryAutoMergeConflict` via the
|
||||
`skipAutoResolveCheck` flag (one place owns the resolveMaxAutoMergeRetries arithmetic). The
|
||||
workspace partial-land decision is `shouldRetryAutoMergeConflict(retries, settings, { skipAutoResolveCheck: true })`.
|
||||
*/
|
||||
const shouldRetryWorkspacePartialLand = (
|
||||
currentRetries: number,
|
||||
settings: { maxAutoMergeRetries?: unknown } | null | undefined,
|
||||
) => shouldRetryAutoMergeConflict(currentRetries, settings, { skipAutoResolveCheck: true });
|
||||
import { createWorkspaceFixture, hasGit, type WorkspaceFixture } from "./_workspace-fixture.js";
|
||||
|
||||
const describeIfGit = hasGit ? describe : describe.skip;
|
||||
@@ -314,6 +325,107 @@ describeIfGit("landWorkspaceTask — landed predicate + finalize-once + idempote
|
||||
});
|
||||
});
|
||||
|
||||
/*
|
||||
FNXC:Workspace 2026-06-22-04:10 (Phase C review A1/A4/A5 — DB-failure resilience):
|
||||
These drive the REAL `landWorkspaceTask` against the REAL two-repo fixture but inject a
|
||||
store whose `updateTask` REJECTS on a chosen patch, exercising the persist-failure windows
|
||||
that the review fixes close. No mock-the-world: the git lands are real; only the targeted
|
||||
DB write is forced to fail.
|
||||
*/
|
||||
describeIfGit("landWorkspaceTask — DB-failure resilience (Phase C review A1/A4/A5)", () => {
|
||||
let fx: WorkspaceFixture;
|
||||
afterEach(() => fx?.cleanup());
|
||||
|
||||
it("A1/A4: a persist-failure AFTER the ref advanced escalates to WorkspacePartialLandError (no silent continue); a retry skips the actually-landed repo (no double squash)", async () => {
|
||||
fx = await createWorkspaceFixture(["repo-a"]);
|
||||
addRepoBranchWithEdit(fx, "repo-a", "a feature\n");
|
||||
const task = makeTask({ "repo-a": { worktreePath: fx.repoPath("repo-a"), branch: BRANCH } });
|
||||
|
||||
// A store that FAILS the landedSha persist (the workspaceWorktrees write) exactly once,
|
||||
// then persists normally — simulating a transient DB hiccup in the A1 window.
|
||||
let failLandedShaWrite = true;
|
||||
const store = createStore(task);
|
||||
const realUpdate = store.updateTask as unknown as (id: string, patch: Partial<Task>) => Promise<undefined>;
|
||||
(store as { updateTask: unknown }).updateTask = vi.fn(async (id: string, patch: Partial<Task>) => {
|
||||
if (failLandedShaWrite && patch.workspaceWorktrees) {
|
||||
failLandedShaWrite = false;
|
||||
throw new Error("synthetic DB write failure (landedSha persist)");
|
||||
}
|
||||
return realUpdate(id, patch);
|
||||
});
|
||||
|
||||
const tipBefore = fx.git("repo-a", "git rev-parse refs/heads/main");
|
||||
|
||||
// First run: repo-a squashes + advances the ref, but the landedSha persist throws.
|
||||
await expect(
|
||||
landWorkspaceTask(store, store.task, fx.rootDir, {}, {
|
||||
mergeAgent: squashMergeAgent(BRANCH),
|
||||
reviewAgent: approveReviewAgent,
|
||||
}),
|
||||
).rejects.toBeInstanceOf(WorkspacePartialLandError);
|
||||
|
||||
// The ref DID advance (the repo is actually landed) — but landedSha was NOT recorded.
|
||||
const tipAfterFirst = fx.git("repo-a", "git rev-parse refs/heads/main");
|
||||
expect(tipAfterFirst).not.toBe(tipBefore);
|
||||
expect(store.task.workspaceWorktrees!["repo-a"].landedSha).toBeUndefined();
|
||||
// Not finalized to done (the throw aborted before finalize).
|
||||
expect(store.moveTaskCalls).toHaveLength(0);
|
||||
// Status was reset off 'merging' before the throw escaped (A3).
|
||||
expect(store.task.status ?? null).toBeNull();
|
||||
|
||||
// Retry: isRepoLanded's trailer ancestor-fallback (A1) recognises the actually-landed
|
||||
// repo via its Fusion-Task-Id trailer and SKIPS it — the ref must NOT advance a 2nd time.
|
||||
const second = await landWorkspaceTask(store, store.task, fx.rootDir, {}, {
|
||||
mergeAgent: squashMergeAgent(BRANCH),
|
||||
reviewAgent: approveReviewAgent,
|
||||
});
|
||||
expect(fx.git("repo-a", "git rev-parse refs/heads/main")).toBe(tipAfterFirst); // no double squash
|
||||
expect(second.repos[0].alreadyLanded).toBe(true);
|
||||
expect(second.allLanded).toBe(true);
|
||||
expect(second.finalized).toBe(true);
|
||||
});
|
||||
|
||||
it("A4: WorkspacePartialLandError is a real class (instanceof + retryable + payload)", () => {
|
||||
const err = new WorkspacePartialLandError(2, ["repo-b"], "partial");
|
||||
expect(err).toBeInstanceOf(WorkspacePartialLandError);
|
||||
expect(err).toBeInstanceOf(Error);
|
||||
expect(err.name).toBe("WorkspacePartialLandError");
|
||||
expect(err.retryable).toBe(true);
|
||||
expect(err.landedCount).toBe(2);
|
||||
expect(err.failedRepos).toEqual(["repo-b"]);
|
||||
});
|
||||
|
||||
it("A5: a rejecting mergeDetails persist aborts finalization (does NOT silently finalize on a stale row)", async () => {
|
||||
fx = await createWorkspaceFixture(["repo-a"]);
|
||||
addRepoBranchWithEdit(fx, "repo-a", "a feature\n");
|
||||
const task = makeTask({ "repo-a": { worktreePath: fx.repoPath("repo-a"), branch: BRANCH } });
|
||||
|
||||
// Fail the mergeDetails write (the finalize TOCTOU window) — the landedSha write succeeds.
|
||||
const store = createStore(task);
|
||||
const realUpdate = store.updateTask as unknown as (id: string, patch: Partial<Task>) => Promise<undefined>;
|
||||
(store as { updateTask: unknown }).updateTask = vi.fn(async (id: string, patch: Partial<Task>) => {
|
||||
if (patch.mergeDetails) {
|
||||
throw new Error("synthetic DB write failure (mergeDetails)");
|
||||
}
|
||||
return realUpdate(id, patch);
|
||||
});
|
||||
|
||||
await expect(
|
||||
landWorkspaceTask(store, store.task, fx.rootDir, {}, {
|
||||
mergeAgent: squashMergeAgent(BRANCH),
|
||||
reviewAgent: approveReviewAgent,
|
||||
}),
|
||||
).rejects.toThrow(/mergeDetails/);
|
||||
|
||||
// Finalization aborted: the task was NOT moved done and no task:merged was emitted on a
|
||||
// stale/unpersisted row.
|
||||
expect(store.moveTaskCalls).toHaveLength(0);
|
||||
expect(store.emitted.some((e) => e.event === "task:merged")).toBe(false);
|
||||
// Status was still reset off 'merging' (A3 finally runs before finalize).
|
||||
expect(store.task.status ?? null).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
describe("workspace partial-land retry/park decision (engine seam, fake timers)", () => {
|
||||
beforeEach(() => vi.useFakeTimers());
|
||||
afterAll(() => vi.useRealTimers());
|
||||
|
||||
@@ -269,4 +269,50 @@ describeIfGit("landWorkspaceTask — per-repo land lease (Phase C U3, KTD4)", ()
|
||||
expect(retry.repos[0].status).toBe("landed");
|
||||
expect(activeSessionRegistry.lookupByPath(repoAbs)).toBeNull();
|
||||
});
|
||||
|
||||
/*
|
||||
FNXC:Workspace 2026-06-22-04:10 (Phase C review A2 — taskId-aware lease across kinds):
|
||||
A FOREIGN-task holder of ANY kind on the sub-repo path is contention for the land
|
||||
busy-check — not only a "workspace-repo-land" holder. Here an EXECUTING task's
|
||||
"workspace-repo-acquire" entry sits on the path; a MERGING task's land must FAST-FAIL
|
||||
with WorkspaceRepoLandBusyError and must NOT clobber the foreign entry.
|
||||
*/
|
||||
it("a foreign-task acquire-lease holder is land contention (busy error) and is NOT clobbered", async () => {
|
||||
fx = await createWorkspaceFixture(["repo-a"]);
|
||||
addRepoBranchWithEdit(fx, "repo-a", "FN-3001", "a feature\n");
|
||||
const repoAbs = fx.repoPath("repo-a");
|
||||
|
||||
// An EXECUTING task (FN-9001) holds an acquire lease on the shared sub-repo path.
|
||||
activeSessionRegistry.registerPath(repoAbs, {
|
||||
taskId: "FN-9001",
|
||||
kind: "workspace-repo-acquire",
|
||||
ownerKey: "workspace-repo-acquire",
|
||||
});
|
||||
const tipBefore = fx.git("repo-a", "git rev-parse refs/heads/main");
|
||||
|
||||
// The MERGING task (FN-3001) tries to land the SAME sub-repo.
|
||||
const task = makeTask("FN-3001", { "repo-a": { worktreePath: repoAbs, branch: BRANCH } });
|
||||
const store = createStore(task);
|
||||
|
||||
let landError: unknown;
|
||||
try {
|
||||
await landWorkspaceTask(store, store.task, fx.rootDir, {}, {
|
||||
mergeAgent: squashMergeAgent(BRANCH),
|
||||
reviewAgent: approveReviewAgent,
|
||||
});
|
||||
} catch (err) {
|
||||
landError = err;
|
||||
}
|
||||
|
||||
// Fast-failed with the retryable busy error — even though the holder kind differs.
|
||||
expect(landError).toBeInstanceOf(WorkspaceRepoLandBusyError);
|
||||
expect((landError as WorkspaceRepoLandBusyError).holderTaskId).toBe("FN-9001");
|
||||
// The foreign acquire entry was NOT clobbered — still owned by FN-9001, same kind.
|
||||
const stillHeld = activeSessionRegistry.lookupByPath(repoAbs);
|
||||
expect(stillHeld?.taskId).toBe("FN-9001");
|
||||
expect(stillHeld?.kind).toBe("workspace-repo-acquire");
|
||||
// The merging task advanced NOTHING and its status was reset off 'merging' (A3).
|
||||
expect(fx.git("repo-a", "git rev-parse refs/heads/main")).toBe(tipBefore);
|
||||
expect(store.task.status ?? null).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -56,12 +56,46 @@ export type SelfOwnedReconcileOutcome =
|
||||
*/
|
||||
export const DEFAULT_SELF_OWNED_MIN_IDLE_MS = 5000;
|
||||
|
||||
/*
|
||||
FNXC:Workspace 2026-06-22-04:10 (Phase C review A2):
|
||||
Thrown by registerPath when a register would overwrite an entry held by a DIFFERENT
|
||||
task on the same path. Surfacing this (rather than silently clobbering) is what stops a
|
||||
merging task's land lease from yanking an executing task's acquire lease on a shared
|
||||
sub-repo. Same-task re-registration is allowed and never throws.
|
||||
*/
|
||||
export class ActiveSessionPathHeldByForeignTaskError extends Error {
|
||||
constructor(
|
||||
public readonly path: string,
|
||||
public readonly holderTaskId: string,
|
||||
public readonly requestingTaskId: string,
|
||||
) {
|
||||
super(
|
||||
`active-session path ${path} is held by task ${holderTaskId}; task ${requestingTaskId} may not overwrite it`,
|
||||
);
|
||||
this.name = "ActiveSessionPathHeldByForeignTaskError";
|
||||
}
|
||||
}
|
||||
|
||||
export class ActiveSessionRegistry {
|
||||
private readonly records = new Map<string, ActiveSessionRecord>();
|
||||
|
||||
/*
|
||||
FNXC:Workspace 2026-06-22-04:10 (Phase C review A2 — taskId-aware lease across kinds):
|
||||
registerPath previously OVERWROTE any existing entry on the path (only console.warn).
|
||||
Because the land lease ("workspace-repo-land") and the execution acquire lease
|
||||
("workspace-repo-acquire") key the SAME sub-repo absolute path, an overwrite let a
|
||||
MERGING task clobber an EXECUTING task's acquire-lease on a shared sub-repo (cross-phase
|
||||
clobber). We now REJECT a register that would overwrite an entry held by a DIFFERENT
|
||||
taskId — regardless of kind — by throwing. Only the SAME task may re-register its own
|
||||
path (idempotent re-registration stays working; this is how an executor re-claims/refreshes
|
||||
its own entry). Callers that may contend (the land lease) must lookupByPath-then-throw a
|
||||
domain busy error BEFORE calling registerPath so they surface contention as a retryable
|
||||
condition rather than this raw guard throw; this guard is the last-line safety net.
|
||||
*/
|
||||
registerPath(worktreePath: string, registration: ActiveSessionRegistration): void {
|
||||
if (this.records.has(worktreePath)) {
|
||||
console.warn(`[active-session-registry] overwriting existing registration for ${worktreePath}`);
|
||||
const existing = this.records.get(worktreePath);
|
||||
if (existing && existing.taskId !== registration.taskId) {
|
||||
throw new ActiveSessionPathHeldByForeignTaskError(worktreePath, existing.taskId, registration.taskId);
|
||||
}
|
||||
this.records.set(worktreePath, {
|
||||
...registration,
|
||||
|
||||
@@ -195,6 +195,13 @@ export { runAiMerge } from "./merger-ai.js";
|
||||
export {
|
||||
landWorkspaceTask,
|
||||
landOneRepo,
|
||||
// FNXC:Workspace 2026-06-22-04:10 (Phase C review A6): canonical landed predicate,
|
||||
// re-exported so Phase D self-healing reuses it instead of reimplementing the ancestor check.
|
||||
isRepoLanded,
|
||||
// FNXC:Workspace 2026-06-22-04:10 (Phase C review A4): real error classes (instanceof-able),
|
||||
// re-exported so the engine dispatch can switch to instanceof in the separate pass.
|
||||
WorkspaceRepoLandBusyError,
|
||||
WorkspacePartialLandError,
|
||||
type WorkspaceMergeResult,
|
||||
type WorkspaceRepoLandResult,
|
||||
type LandOneRepoResult,
|
||||
|
||||
@@ -99,6 +99,19 @@ async function gitOk(args: string[], cwd: string): Promise<boolean> {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* FNXC:Workspace 2026-06-22-04:10 (Phase C review A1):
|
||||
* Capture git stdout, returning undefined (never throwing) on failure — for read-only
|
||||
* probes (merge-base, log --grep) where a non-zero exit is an expected "not found".
|
||||
*/
|
||||
async function gitCapture(args: string[], cwd: string): Promise<string | undefined> {
|
||||
try {
|
||||
return await git(args, cwd);
|
||||
} catch {
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
|
||||
function getErrorMessage(err: unknown): string {
|
||||
return err instanceof Error ? err.message : String(err);
|
||||
}
|
||||
@@ -1445,6 +1458,34 @@ export class WorkspaceRepoLandBusyError extends Error {
|
||||
}
|
||||
}
|
||||
|
||||
/*
|
||||
FNXC:Workspace 2026-06-22-04:10 (Phase C review A4 — real WorkspacePartialLandError class):
|
||||
Previously the partial-land signal was a bare `new Error()` with `.name` patched in
|
||||
project-engine.ts (a footgun: no instanceof, no typed payload). It is now a real exported
|
||||
class so the dispatch can switch to `instanceof` (separate pass) and tests can assert
|
||||
`instanceof`. `retryable = true` because a partial land is recoverable — the landed repos'
|
||||
`landedSha` is persisted and a re-run skips them (the U2 idempotency contract).
|
||||
|
||||
`landWorkspaceTask` throws this from ONE place: the A1 persist-after-advance failure window
|
||||
(the integration ref ALREADY advanced but `persistRepoLandedSha` could not record the
|
||||
`landedSha`). The ORDINARY partial land (repo A landed, repo B's land failed) still RETURNS
|
||||
`allLanded:false` — that return-based contract is what the engine dispatch and the oracle
|
||||
workspace-merger tests already consume; only the persist-failure window escalates to a throw
|
||||
so the engine parks/retries and A1's `isRepoLanded` ancestor-fallback skips the actually-landed
|
||||
repo on retry (no double-squash).
|
||||
*/
|
||||
export class WorkspacePartialLandError extends Error {
|
||||
public readonly retryable = true;
|
||||
constructor(
|
||||
public readonly landedCount: number,
|
||||
public readonly failedRepos: string[],
|
||||
message: string,
|
||||
) {
|
||||
super(message);
|
||||
this.name = "WorkspacePartialLandError";
|
||||
}
|
||||
}
|
||||
|
||||
export async function landWorkspaceTask(
|
||||
store: TaskStore,
|
||||
task: Task,
|
||||
@@ -1482,6 +1523,18 @@ export async function landWorkspaceTask(
|
||||
let allLanded = true;
|
||||
|
||||
await setStatus("merging");
|
||||
/*
|
||||
FNXC:Workspace 2026-06-22-04:10 (Phase C review A3 — status 'merging' must never leak):
|
||||
The busy-throw (WorkspaceRepoLandBusyError) and the persist-failure throw
|
||||
(WorkspacePartialLandError) exit the loop BEFORE the post-loop `setStatus(null)`. If the
|
||||
engine catch never runs (process crash between throw and catch) the task stays stuck
|
||||
'merging' with no manual door to clear it. Wrap the whole per-repo loop so `setStatus(null)`
|
||||
ALWAYS runs (in finally) before ANY throw escapes. The success path still finalizes to done
|
||||
AFTER this finally (finalizeWorkspaceTask sets its own column/status), so clearing 'merging'
|
||||
first is safe — finalize overwrites it. This finally only clears the transient merge status;
|
||||
it does not move the task.
|
||||
*/
|
||||
try {
|
||||
for (const repoRel of repoKeys) {
|
||||
throwIfAborted(options.signal, taskId);
|
||||
const entry = workspaceWorktrees[repoRel];
|
||||
@@ -1508,7 +1561,7 @@ export async function landWorkspaceTask(
|
||||
// ancestor of (or equals) its CURRENT integration tip is already landed — SKIP
|
||||
// it so a retry never re-advances the ref. This makes a re-run after a partial
|
||||
// land idempotent for the already-landed repos.
|
||||
if (await isRepoLanded(repoRootDir, integrationBranch, entry.landedSha)) {
|
||||
if (await isRepoLanded(repoRootDir, integrationBranch, entry.landedSha, taskId, entry.branch)) {
|
||||
await log(`AI merge (workspace): sub-repo ${repoRel} already landed (${short(entry.landedSha!)} ⊑ ${integrationBranch}) — skipping`);
|
||||
repos.push({
|
||||
repo: repoRel, repoRootDir, integrationBranch, branch: entry.branch,
|
||||
@@ -1526,15 +1579,19 @@ export async function landWorkspaceTask(
|
||||
interleaved await would let a second task pass the gate before we register. If
|
||||
another task holds the land lease we FAST-FAIL with a retryable busy error; the
|
||||
U2 dispatch auto-retry/park path handles it (no waiting lock reimplemented here).
|
||||
We only treat a HELD entry of OUR OWN land ownerKey as contention, so a stale
|
||||
entry of a different kind on this path (e.g. a leftover acquire entry) is ignored.
|
||||
|
||||
FNXC:Workspace 2026-06-22-04:10 (Phase C review A2 — taskId-aware contention across kinds):
|
||||
Previously we only treated a HELD entry of OUR OWN land ownerKey as contention, so a
|
||||
MERGING task would registerPath-OVERWRITE an EXECUTING task's "workspace-repo-acquire"
|
||||
entry on a shared sub-repo (cross-phase clobber). Now ANY foreign-task holder on this
|
||||
path — regardless of kind (acquire OR land OR anything else) — is contention: we throw
|
||||
WorkspaceRepoLandBusyError so the engine retries when the other task releases its hold.
|
||||
A SAME-task holder is NOT contention (idempotent re-claim of our own path). The
|
||||
registerPath guard (A2b) backstops this: it also rejects a foreign-task overwrite, so a
|
||||
missed check can never silently clobber.
|
||||
*/
|
||||
const landLeaseHolder = activeSessionRegistry.lookupByPath(repoRootDir);
|
||||
if (
|
||||
landLeaseHolder &&
|
||||
landLeaseHolder.ownerKey === WORKSPACE_REPO_LAND_OWNER_KEY &&
|
||||
landLeaseHolder.taskId !== taskId
|
||||
) {
|
||||
if (landLeaseHolder && landLeaseHolder.taskId !== taskId) {
|
||||
throw new WorkspaceRepoLandBusyError(repoRel, landLeaseHolder.taskId, taskId);
|
||||
}
|
||||
activeSessionRegistry.registerPath(repoRootDir, {
|
||||
@@ -1551,10 +1608,32 @@ export async function landWorkspaceTask(
|
||||
allowDirtyLocalCheckoutSync: options.allowDirtyLocalCheckoutSync === true,
|
||||
});
|
||||
if (landResult.outcome === "landed") {
|
||||
// Persist this repo's landedSha BEFORE moving on (fresh-read-then-merge so
|
||||
// sibling entries written by a concurrent path are not clobbered). The retry
|
||||
// predicate above reads this back to skip the repo on a re-run.
|
||||
await persistRepoLandedSha(store, taskId, repoRel, landResult.squashSha);
|
||||
/*
|
||||
FNXC:Workspace 2026-06-22-04:10 (Phase C review A1 — persist-after-advance is a HARD failure):
|
||||
The integration ref has ALREADY advanced (squash landed) by the time we persist
|
||||
`landedSha`. If the DB write fails here the ref is advanced but UNRECORDED — we must NOT
|
||||
silently continue (a return-based partial would let a retry double-squash). Escalate to a
|
||||
retryable WorkspacePartialLandError so the engine parks/retries; on retry, `isRepoLanded`'s
|
||||
trailer ancestor-fallback recognises this actually-landed repo and skips it. The repo IS
|
||||
recorded as `landed` in the in-memory result first so the error payload is accurate.
|
||||
*/
|
||||
try {
|
||||
await persistRepoLandedSha(store, taskId, repoRel, landResult.squashSha);
|
||||
} catch (persistErr: unknown) {
|
||||
const pmsg = getErrorMessage(persistErr);
|
||||
await log(`AI merge (workspace): sub-repo ${repoRel} landed (${short(landResult.squashSha)}) but persisting landedSha FAILED: ${pmsg} — escalating to partial land so a retry can recover (ref already advanced; retry will skip via trailer ancestor-check)`);
|
||||
repos.push({
|
||||
repo: repoRel, repoRootDir, integrationBranch, branch: entry.branch,
|
||||
status: "landed", landedSha: landResult.squashSha, localSync: landResult.localSync,
|
||||
});
|
||||
allLanded = false;
|
||||
const landedCount = repos.filter((r) => r.status === "landed").length;
|
||||
throw new WorkspacePartialLandError(
|
||||
landedCount,
|
||||
[repoRel],
|
||||
`Workspace land for ${taskId}: sub-repo ${repoRel} advanced its integration ref but the landedSha persist failed (${pmsg}); retry to record/skip it`,
|
||||
);
|
||||
}
|
||||
repos.push({
|
||||
repo: repoRel, repoRootDir, integrationBranch, branch: entry.branch,
|
||||
status: "landed", landedSha: landResult.squashSha, localSync: landResult.localSync,
|
||||
@@ -1563,6 +1642,9 @@ export async function landWorkspaceTask(
|
||||
repos.push({ repo: repoRel, repoRootDir, integrationBranch, branch: entry.branch, status: "empty" });
|
||||
}
|
||||
} catch (err: unknown) {
|
||||
// A WorkspacePartialLandError from the persist-failure window above must PROPAGATE
|
||||
// (the engine parks/retries). The outer try/finally below resets status first (A3).
|
||||
if (err instanceof WorkspacePartialLandError) throw err;
|
||||
const message = getErrorMessage(err);
|
||||
await log(`AI merge (workspace): sub-repo ${repoRel} land failed: ${message}`);
|
||||
await audit.git({ type: "merge:ai-no-branch", target: entry.branch, metadata: { taskId, kind: "workspace-repo-land-failed", repo: repoRel, error: message } }).catch(() => undefined);
|
||||
@@ -1586,8 +1668,12 @@ export async function landWorkspaceTask(
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
await setStatus(null);
|
||||
} finally {
|
||||
// A3: clear the transient 'merging' status before ANY throw (busy / partial-land /
|
||||
// abort) escapes, AND on the normal fall-through. The success path's finalize below
|
||||
// re-sets the task's column/status to done, so clearing here first is safe.
|
||||
await setStatus(null);
|
||||
}
|
||||
|
||||
// U2 finalize-once (KTD3): move the task to `done` EXACTLY ONCE, only after EVERY
|
||||
// acquired repo's landed predicate holds (all landed/empty, none failed). Reuse the
|
||||
@@ -1609,18 +1695,65 @@ export async function landWorkspaceTask(
|
||||
* the landed commit is still reachable, so the repo stays "landed". A `landedSha` that
|
||||
* is NOT reachable from the tip (e.g. the ref was reset/rebuilt) reads as NOT landed and
|
||||
* the repo re-lands.
|
||||
*
|
||||
* FNXC:Workspace 2026-06-22-04:10 (Phase C review A1 — task-trailer ancestor fallback):
|
||||
* The double-land window: a land advances the integration ref via `advanceIntegrationBranchRef`'s
|
||||
* CAS, then `persistRepoLandedSha` records `landedSha`. If that DB write fails AFTER the ref
|
||||
* advanced, the repo is ACTUALLY landed but has NO recorded `landedSha`, so the landedSha check
|
||||
* above reports NOT-landed → a retry re-runs `landOneRepo`, the CAS rebuilds, and a SECOND squash
|
||||
* lands (not idempotent). To close the window we ALSO treat the repo as landed when the live
|
||||
* integration ref carries a commit with THIS task's `Fusion-Task-Id` trailer.
|
||||
*
|
||||
* Why a trailer scan and NOT a branch-tip ancestor check: the land is a `git merge --squash`,
|
||||
* whose squash commit's parent is the integration tip, NOT the task branch — so `merge-base
|
||||
* --is-ancestor <branch> <integration>` is FALSE even right after a successful land. The
|
||||
* `Fusion-Task-Id` trailer (always stamped onto the squash by `taskTrailers` + the
|
||||
* ensureTaskMetadata safety net) is the only reliable "this task's work is already on the ref"
|
||||
* signal that does not depend on the landedSha row, so it is what survives a lost persist. We
|
||||
* bound the scan to commits the integration tip has gained since the branch's merge-base (the
|
||||
* land base) so an unrelated historical reuse of the same trailer cannot false-positive.
|
||||
*
|
||||
* Exported (A6) so Phase D self-healing reuses THIS canonical predicate instead of
|
||||
* reimplementing the ancestor/trailer check.
|
||||
*/
|
||||
async function isRepoLanded(
|
||||
export async function isRepoLanded(
|
||||
repoRootDir: string,
|
||||
integrationBranch: string,
|
||||
landedSha: string | undefined,
|
||||
taskId?: string,
|
||||
branch?: string,
|
||||
): Promise<boolean> {
|
||||
if (!landedSha) return false;
|
||||
if (!(await gitOk(["rev-parse", "--verify", `refs/heads/${integrationBranch}`], repoRootDir))) {
|
||||
const intRef = `refs/heads/${integrationBranch}`;
|
||||
if (!(await gitOk(["rev-parse", "--verify", intRef], repoRootDir))) {
|
||||
return false;
|
||||
}
|
||||
// Primary: recorded landedSha is an ancestor of (or equals) the integration tip.
|
||||
// `merge-base --is-ancestor X Y` exits 0 iff X is an ancestor of (or equal to) Y.
|
||||
return await gitOk(["merge-base", "--is-ancestor", landedSha, `refs/heads/${integrationBranch}`], repoRootDir);
|
||||
if (
|
||||
landedSha &&
|
||||
(await gitOk(["merge-base", "--is-ancestor", landedSha, intRef], repoRootDir))
|
||||
) {
|
||||
return true;
|
||||
}
|
||||
// A1 fallback: even without a recorded landedSha, the repo is already landed if the
|
||||
// integration ref carries a commit with this task's Fusion-Task-Id trailer (the squash
|
||||
// we lost the persist for). Bound the scan to commits gained since the branch's land base
|
||||
// so a stale historical trailer of the same id cannot false-positive.
|
||||
if (taskId) {
|
||||
const branchRef = branch ? `refs/heads/${branch}` : undefined;
|
||||
let range = intRef;
|
||||
if (branchRef && (await gitOk(["rev-parse", "--verify", branchRef], repoRootDir))) {
|
||||
const base = await gitCapture(["merge-base", branchRef, intRef], repoRootDir);
|
||||
if (base) range = `${base.trim()}..${intRef}`;
|
||||
}
|
||||
const trailer = `${FUSION_TASK_ID_TRAILER_KEY}: ${taskId}`;
|
||||
const found = await gitCapture(
|
||||
["log", "--format=%H", `--grep=${trailer}`, "--fixed-strings", range],
|
||||
repoRootDir,
|
||||
);
|
||||
if (found && found.trim().length > 0) return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -1628,6 +1761,17 @@ async function isRepoLanded(
|
||||
* 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.
|
||||
*
|
||||
* 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
|
||||
* double-land bug: the integration ref has ALREADY advanced by the time we persist, so a
|
||||
* silently-lost write means `landedSha` is never recorded → on retry the landedSha check sees
|
||||
* NOT-landed and re-runs the squash (a SECOND squash commit). We now PROPAGATE the write
|
||||
* failure. The caller (`landWorkspaceTask`) catches it as a partial-land for this repo and
|
||||
* escalates to `WorkspacePartialLandError` so the engine parks/retries; on retry, `isRepoLanded`'s
|
||||
* 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.
|
||||
*/
|
||||
async function persistRepoLandedSha(
|
||||
store: TaskStore,
|
||||
@@ -1635,12 +1779,12 @@ async function persistRepoLandedSha(
|
||||
repoRel: string,
|
||||
landedSha: string,
|
||||
): Promise<void> {
|
||||
const latest = await store.getTask(taskId).catch(() => undefined);
|
||||
const latest = await store.getTask(taskId);
|
||||
const current = latest?.workspaceWorktrees ?? {};
|
||||
const entry = current[repoRel];
|
||||
if (!entry) return; // entry vanished — nothing to merge into
|
||||
const next = { ...current, [repoRel]: { ...entry, landedSha } };
|
||||
await store.updateTask(taskId, { workspaceWorktrees: next }).catch(() => undefined);
|
||||
await store.updateTask(taskId, { workspaceWorktrees: next });
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -1663,14 +1807,28 @@ async function finalizeWorkspaceTask(
|
||||
const representative = landed.length > 0 ? landed[0].landedSha : undefined;
|
||||
const anyLanded = landed.length > 0;
|
||||
|
||||
// Pre-populate task.mergeDetails so finalizeTask's spread carries the workspace map.
|
||||
/*
|
||||
FNXC:Workspace 2026-06-22-04:10 (Phase C review A5 — fresh-read + no-swallow finalize):
|
||||
Two fixes to the FN-5627 TOCTOU class:
|
||||
1. The `task` argument is the SNAPSHOT captured at the START of `landWorkspaceTask`; by
|
||||
finalize time the persisted row has gained each repo's `landedSha` (and possibly other
|
||||
concurrent edits). Spreading the stale snapshot's mergeDetails could drop/clobber those.
|
||||
Re-read the LATEST task and spread ITS mergeDetails (fresh-read-then-merge), falling back
|
||||
to the snapshot only if the read fails.
|
||||
2. The `store.updateTask(...)` was `.catch(() => undefined)` — a swallowed write left the
|
||||
in-memory `mergeConfirmed:true` while the persisted row stayed stale (the finalize would
|
||||
then report done with an unpersisted merge). PROPAGATE the failure so finalization aborts
|
||||
and self-healing recovers, rather than silently finalizing on a stale row.
|
||||
*/
|
||||
const fresh = await store.getTask(taskId).catch(() => undefined);
|
||||
const baseMergeDetails = fresh?.mergeDetails ?? task.mergeDetails;
|
||||
const mergeDetails: MergeDetails = {
|
||||
...task.mergeDetails,
|
||||
...baseMergeDetails,
|
||||
...(representative ? { commitSha: representative } : {}),
|
||||
...(anyLanded ? { workspaceLandedShas } : {}),
|
||||
mergeConfirmed: anyLanded,
|
||||
};
|
||||
await store.updateTask(taskId, { mergeDetails }).catch(() => undefined);
|
||||
await store.updateTask(taskId, { mergeDetails });
|
||||
task.mergeDetails = mergeDetails;
|
||||
|
||||
const result: MergeResult = {
|
||||
|
||||
@@ -13,7 +13,7 @@ import type {
|
||||
ResearchSynthesisRequest,
|
||||
ResearchSynthesisResult,
|
||||
} from "@fusion/core";
|
||||
import { allowsAutoMergeProcessing, compareTasksByPriorityThenAgeAndId, getTaskHardMergeBlocker, isSharedBranchGroupMemberIntegration, normalizeMergerMode, resolveMaxAutoMergeRetries, sortTasksByPriorityThenAgeAndId } from "@fusion/core";
|
||||
import { allowsAutoMergeProcessing, compareTasksByPriorityThenAgeAndId, getTaskHardMergeBlocker, isSharedBranchGroupMemberIntegration, isWorkspaceTask, normalizeMergerMode, resolveMaxAutoMergeRetries, sortTasksByPriorityThenAgeAndId } from "@fusion/core";
|
||||
import { execFile } from "node:child_process";
|
||||
import { promisify } from "node:util";
|
||||
import { InProcessRuntime } from "./runtimes/in-process-runtime.js";
|
||||
@@ -31,7 +31,7 @@ import { createFusionAuthStorage, getFusionOAuthAlertStatePath } from "./auth-st
|
||||
import { CronRunner, createAiPromptExecutor } from "./cron-runner.js";
|
||||
import type { RoutineRunner } from "./routine-runner.js";
|
||||
import { sweepStaleAutostashes, VerificationError } from "./merger.js";
|
||||
import { runAiMerge, landWorkspaceTask } from "./merger-ai.js";
|
||||
import { runAiMerge, landWorkspaceTask, WorkspacePartialLandError, WorkspaceRepoLandBusyError } from "./merger-ai.js";
|
||||
import { promoteBranchGroup, type BranchGroupPromotionResult, type CreateGroupPrFn, type SyncGroupPrFn } from "./group-merge-coordinator.js";
|
||||
import { PRIORITY_MERGE } from "./concurrency.js";
|
||||
import { runtimeLog } from "./logger.js";
|
||||
@@ -125,35 +125,27 @@ function isInvalidDoneTransitionError(error: unknown): boolean {
|
||||
return message.includes("Invalid transition:") && message.includes("→ 'done'");
|
||||
}
|
||||
|
||||
/*
|
||||
FNXC:Workspace 2026-06-22-05:10 (Phase C review B6 — unify partial-land retry seam):
|
||||
The workspace PARTIAL-land retry decision (some sub-repos landed, one failed) is the SAME
|
||||
arithmetic as the conflict-retry decision MINUS the `autoResolveConflicts` gate (a partial
|
||||
land is retryable regardless of conflict-resolution settings, because the landed repos'
|
||||
`landedSha` is persisted and a re-run skips them — U2 idempotency). To keep the
|
||||
`resolveMaxAutoMergeRetries(settings)` arithmetic in ONE place we collapse the former
|
||||
`shouldRetryWorkspacePartialLand` into this function via `skipAutoResolveCheck`. When set,
|
||||
the `autoResolveConflicts` gate is bypassed; otherwise behavior is byte-identical to before.
|
||||
`currentRetries + 1 < MAX` keeps the LAST attempt's failure parking in the same tick rather
|
||||
than scheduling an Nth timer that a restart could strand.
|
||||
*/
|
||||
export function shouldRetryAutoMergeConflict(
|
||||
currentRetries: number,
|
||||
settings: { autoResolveConflicts?: boolean; maxAutoMergeRetries?: unknown } | null | undefined,
|
||||
opts?: { skipAutoResolveCheck?: boolean },
|
||||
): { shouldRetry: boolean; maxAutoMergeRetries: number; nextRetryCount: number } {
|
||||
const maxAutoMergeRetries = resolveMaxAutoMergeRetries(settings);
|
||||
const autoResolveOk = opts?.skipAutoResolveCheck === true || settings?.autoResolveConflicts !== false;
|
||||
return {
|
||||
shouldRetry: settings?.autoResolveConflicts !== false && currentRetries + 1 < maxAutoMergeRetries,
|
||||
maxAutoMergeRetries,
|
||||
nextRetryCount: currentRetries + 1,
|
||||
};
|
||||
}
|
||||
|
||||
/*
|
||||
FNXC:Workspace 2026-06-22-00:30 (Phase C U2, KTD3):
|
||||
Pure retry/park decision for a workspace PARTIAL land (some sub-repos landed, one failed).
|
||||
Mirrors `shouldRetryAutoMergeConflict` so the engine dispatch's partial-land catch branch
|
||||
has a narrow, unit-testable seam: a partial land is RETRYABLE (the landed repos' `landedSha`
|
||||
is persisted, so a re-run skips them and only the failed repo retries), so it CONSUMES a
|
||||
mergeRetry and re-enqueues up to `resolveMaxAutoMergeRetries(settings)`, then OPERATOR-PARKS
|
||||
(`shouldRetry:false`). `currentRetries + 1 < MAX` keeps the LAST attempt's failure parking
|
||||
in the same tick rather than scheduling an Nth timer that a restart could strand.
|
||||
*/
|
||||
export function shouldRetryWorkspacePartialLand(
|
||||
currentRetries: number,
|
||||
settings: { maxAutoMergeRetries?: unknown } | null | undefined,
|
||||
): { shouldRetry: boolean; maxAutoMergeRetries: number; nextRetryCount: number } {
|
||||
const maxAutoMergeRetries = resolveMaxAutoMergeRetries(settings);
|
||||
return {
|
||||
shouldRetry: currentRetries + 1 < maxAutoMergeRetries,
|
||||
shouldRetry: autoResolveOk && currentRetries + 1 < maxAutoMergeRetries,
|
||||
maxAutoMergeRetries,
|
||||
nextRetryCount: currentRetries + 1,
|
||||
};
|
||||
@@ -370,6 +362,19 @@ export class ProjectEngine {
|
||||
private autostashSweepTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
private mergeActiveReconcileTimer: ReturnType<typeof setInterval> | null = null;
|
||||
|
||||
/*
|
||||
FNXC:Workspace 2026-06-22-05:10 (Phase C review B4 — separate busy-retry quota):
|
||||
Transient sub-repo land-lease contention (WorkspaceRepoLandBusyError) must NOT burn the
|
||||
persisted `mergeRetries` quota — two tasks contending for the same sub-repo could otherwise
|
||||
exhaust all retries on pure busy-errors before a single real land attempt, then park a
|
||||
never-failed task. We track busy re-enqueues in this in-memory, per-task counter (transient
|
||||
contention need not survive a restart) and CAP it separately from `mergeRetries`. A real
|
||||
partial land (WorkspacePartialLandError) still consumes `mergeRetries` up to MAX, then parks.
|
||||
Cleared on the first non-busy outcome (success path resets it).
|
||||
*/
|
||||
private workspaceBusyReenqueues = new Map<string, number>();
|
||||
private static readonly WORKSPACE_BUSY_MAX_REENQUEUES = 10;
|
||||
|
||||
/**
|
||||
* Pending manual merge resolvers — keyed by taskId.
|
||||
* When `onMerge` is called, the task is enqueued like auto-merge but a
|
||||
@@ -1866,6 +1871,19 @@ export class ProjectEngine {
|
||||
// in-review by auto-recovery after a successful merge) — just
|
||||
// complete the task without re-running the merge process.
|
||||
if (task.mergeDetails?.mergeConfirmed) {
|
||||
/*
|
||||
FNXC:Workspace 2026-06-22-05:10 (Phase C review B2 — fast-path must skip workspace tasks):
|
||||
The FN-5627 reachability gate below runs `git cat-file -e <commitSha>` in cwd = the
|
||||
project/workspace ROOT. For a WORKSPACE task, `finalizeWorkspaceTask` records
|
||||
`mergeDetails.commitSha` = the FIRST sorted sub-repo's squash sha, which lives in
|
||||
`join(workspaceRoot, <repo>)`, NOT in the workspace root (which is not even a git repo).
|
||||
So `cat-file -e` against the root cwd ALWAYS reports commit-missing → the gate would
|
||||
clear `mergeConfirmed` and demote/park a FULLY-MERGED workspace task. Workspace tasks
|
||||
are merge-verified by each sub-repo's persisted `landedSha`, not a single root-cwd
|
||||
commitSha, so the root-cwd reachability gate does not apply to them. SKIP the gate for
|
||||
workspace tasks and take the fast-path. (Per-sub-repo cwd reachability verification is a
|
||||
larger change deferred past Phase C; skipping here is the correct minimal fix.)
|
||||
*/
|
||||
// FN-5627: Reachability defense-in-depth. The merger has a TOCTOU
|
||||
// window where `mergeConfirmed: true` can be persisted to the task
|
||||
// row before `git update-ref refs/heads/<integration>` actually
|
||||
@@ -1890,6 +1908,7 @@ export class ProjectEngine {
|
||||
`Auto-merge: ${taskId} merge-confirmed fast-path rerouting shared-group member from ${task.mergeDetails.mergeTargetBranch} to ${routedFastPathTarget}`,
|
||||
);
|
||||
}
|
||||
if (!isWorkspaceTask(task)) {
|
||||
const reachability = await verifyMergeConfirmedReachability({
|
||||
commitSha: task.mergeDetails.commitSha,
|
||||
integrationBranch: integrationBranchForGate,
|
||||
@@ -2032,6 +2051,7 @@ export class ProjectEngine {
|
||||
this.internalEnqueueMerge(taskId);
|
||||
continue;
|
||||
}
|
||||
} // end !isWorkspaceTask reachability gate (B2): workspace tasks skip the root-cwd commitSha check
|
||||
const blockerReason = getTaskHardMergeBlocker({
|
||||
...(task as Task),
|
||||
// Merge-confirmed tasks have already landed. Treat stale merge
|
||||
@@ -2320,8 +2340,7 @@ export class ProjectEngine {
|
||||
// routing falls through to runAiMerge, whose chokepoint guard re-reads
|
||||
// the task and is the authoritative workspace enforcement.
|
||||
const mergeTask = await store.getTask(taskId).catch(() => null);
|
||||
const isWorkspaceMerge =
|
||||
!!mergeTask?.workspaceWorktrees && Object.keys(mergeTask.workspaceWorktrees).length > 0;
|
||||
const isWorkspaceMerge = !!mergeTask && isWorkspaceTask(mergeTask);
|
||||
if (isWorkspaceMerge) {
|
||||
// FNXC:Workspace 2026-06-22-00:30 (Phase C U2, KTD3):
|
||||
// Land each acquired sub-repo on its own local integration ref;
|
||||
@@ -2339,14 +2358,18 @@ export class ProjectEngine {
|
||||
{ ...mergerOptions, allowDirtyLocalCheckoutSync: settings.merger?.allowDirtyLocalCheckoutSync === true },
|
||||
);
|
||||
if (!workspaceResult.allLanded) {
|
||||
// FNXC:Workspace 2026-06-22-05:10 (Phase C review B7):
|
||||
// Throw the real exported WorkspacePartialLandError class (not a bare Error with
|
||||
// a patched `.name`) so the catch below can match via `instanceof` and read the
|
||||
// typed payload (landedCount, failedRepos).
|
||||
const failed = workspaceResult.repos.filter((r) => r.status === "failed");
|
||||
const landedCount = workspaceResult.repos.filter((r) => r.status === "landed").length;
|
||||
const detail = failed.map((r) => `${r.repo}: ${r.error ?? "land failed"}`).join("; ");
|
||||
const partialErr = new Error(
|
||||
throw new WorkspacePartialLandError(
|
||||
landedCount,
|
||||
failed.map((r) => r.repo),
|
||||
`Workspace partial land for ${taskId}: ${landedCount} repo(s) landed, ${failed.length} failed — ${detail}`,
|
||||
);
|
||||
partialErr.name = "WorkspacePartialLandError";
|
||||
throw partialErr;
|
||||
}
|
||||
// Finalized to done by landWorkspaceTask; report the merge as merged so
|
||||
// the success path (retry reset + branch-group promotion) runs normally.
|
||||
@@ -2409,6 +2432,9 @@ export class ProjectEngine {
|
||||
if (latestTask?.mergeRetries && latestTask.mergeRetries > 0) {
|
||||
await store.updateTask(taskId, { mergeRetries: 0 });
|
||||
}
|
||||
// FNXC:Workspace 2026-06-22-05:10 (Phase C review B4): clear the in-memory busy
|
||||
// re-enqueue counter once the merge succeeds so a later unrelated contention starts fresh.
|
||||
this.workspaceBusyReenqueues.delete(taskId);
|
||||
|
||||
await attemptBranchGroupPromotion(latestTask);
|
||||
}
|
||||
@@ -2460,40 +2486,98 @@ export class ProjectEngine {
|
||||
continue;
|
||||
}
|
||||
|
||||
/*
|
||||
FNXC:Workspace 2026-06-22-05:10 (Phase C review B4/B7 — busy contention split from real partial land):
|
||||
A `WorkspaceRepoLandBusyError` (a second task holds the same sub-repo's land lease) is
|
||||
TRANSIENT contention, not a land failure: re-enqueue it with backoff WITHOUT consuming the
|
||||
persisted `mergeRetries` quota, bounded separately by `workspaceBusyReenqueues`
|
||||
(WORKSPACE_BUSY_MAX_REENQUEUES). This stops two contending tasks from exhausting all merge
|
||||
retries on busy-errors before either makes a real land attempt, then parking a never-failed
|
||||
task. Detect via `instanceof` now that both are exported classes (B7).
|
||||
*/
|
||||
if (err instanceof WorkspaceRepoLandBusyError && !hasManualResolver) {
|
||||
const busyCount = this.workspaceBusyReenqueues.get(taskId) ?? 0;
|
||||
await store
|
||||
.logEntry(taskId, `Workspace sub-repo land busy (contention): ${errorMsg}`, "WorkspaceRepoLandBusy")
|
||||
.catch(() => undefined);
|
||||
if (busyCount < ProjectEngine.WORKSPACE_BUSY_MAX_REENQUEUES) {
|
||||
this.workspaceBusyReenqueues.set(taskId, busyCount + 1);
|
||||
// Capped exponential backoff (B5): never exceed 60s even at the busy ceiling.
|
||||
const delayMs = Math.min(5000 * Math.pow(2, busyCount), 60_000);
|
||||
await store.updateTask(taskId, { status: null }).catch(() => undefined);
|
||||
runtimeLog.log(
|
||||
`Workspace land busy re-enqueue ${busyCount + 1}/${ProjectEngine.WORKSPACE_BUSY_MAX_REENQUEUES} for ${taskId} in ${delayMs / 1000}s (no mergeRetry consumed — pure lease contention)`,
|
||||
);
|
||||
setTimeout(() => {
|
||||
if (!this.shuttingDown) this.internalEnqueueMerge(taskId);
|
||||
}, delayMs);
|
||||
} else {
|
||||
// Pathological sustained contention — surface but do NOT burn mergeRetries; park as
|
||||
// failed so the cooldown sweep stops re-attempting and an operator can intervene.
|
||||
this.workspaceBusyReenqueues.delete(taskId);
|
||||
await store
|
||||
.updateTask(taskId, { status: "failed", error: errorMsg })
|
||||
.catch(() => undefined);
|
||||
runtimeLog.error(
|
||||
`Auto-merge: ${taskId} workspace land busy ${ProjectEngine.WORKSPACE_BUSY_MAX_REENQUEUES} times — parked as failed (sustained sub-repo lease contention)`,
|
||||
);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
// FNXC:Workspace 2026-06-22-00:30 (Phase C U2, KTD3):
|
||||
// Workspace PARTIAL-LAND auto-retry-then-park (user decision). Unlike the R7
|
||||
// WorkspaceTaskMergeError above (a permanent config error that must NOT burn
|
||||
// retries), a partial land — repo A landed, repo B failed — is RETRYABLE: the
|
||||
// landed repos' `landedSha` is persisted, so a re-run of `landWorkspaceTask`
|
||||
// skips them and re-attempts only the failed repo (idempotent). So this CONSUMES
|
||||
// a `mergeRetry` and re-enqueues the merge with exponential backoff up to the
|
||||
// a `mergeRetry` and re-enqueues the merge with capped exponential backoff up to the
|
||||
// existing MAX (resolveMaxAutoMergeRetries), then OPERATOR-PARKS (status:"failed")
|
||||
// — mirroring the conflict-retry seam below. Detect by err.name (robust across
|
||||
// the package boundary). Manual merges fall through to rejectMergeResolvers at
|
||||
// the hasManualResolver early-return below (no auto-retry for manual).
|
||||
/*
|
||||
FNXC:Workspace 2026-06-22-02:10 (Phase C U3, KTD4):
|
||||
A `WorkspaceRepoLandBusyError` (a second task holds the same sub-repo's
|
||||
land lease) is ALSO retryable here — it is transient contention, not a
|
||||
terminal failure. Route it through the SAME auto-retry-then-park seam (it
|
||||
consumes a mergeRetry and re-enqueues with backoff; a re-run skips
|
||||
already-landed repos and finds the lease freed). Detect by err.name across
|
||||
the package boundary, same as the partial-land error.
|
||||
*/
|
||||
const isWorkspacePartialLand =
|
||||
err instanceof Error &&
|
||||
(err.name === "WorkspacePartialLandError" || err.name === "WorkspaceRepoLandBusyError");
|
||||
if (isWorkspacePartialLand && !hasManualResolver) {
|
||||
const wsSettings = await store.getSettings().catch(() => ({ maxAutoMergeRetries: undefined }));
|
||||
// — reusing the unified shouldRetryAutoMergeConflict seam with skipAutoResolveCheck
|
||||
// (B6). Detect via `instanceof` (B7). Manual merges fall through to
|
||||
// rejectMergeResolvers at the hasManualResolver early-return below.
|
||||
if (err instanceof WorkspacePartialLandError && !hasManualResolver) {
|
||||
const wsSettings = await store.getSettings().catch(() => null);
|
||||
const wsTask = await store.getTask(taskId).catch(() => null);
|
||||
const wsRetries = wsTask?.mergeRetries ?? 0;
|
||||
const decision = shouldRetryWorkspacePartialLand(wsRetries, wsSettings as { maxAutoMergeRetries?: unknown });
|
||||
/*
|
||||
FNXC:Workspace 2026-06-22-05:10 (Phase C review B1 — fail closed on getTask null):
|
||||
If getTask returns null (DB outage), we CANNOT read `mergeRetries`. Defaulting to 0
|
||||
would make `shouldRetry` always true while the increment updateTask also fails against
|
||||
the non-responsive DB → an indefinite setTimeout retry storm against a dead DB. FAIL
|
||||
CLOSED: do not schedule a retry. Attempt a best-effort park to `failed`; if that write
|
||||
also fails it throws away cleanly and the cooldown sweep (canMergeTask) will re-evaluate
|
||||
once the DB recovers, rather than hammering it on a tight timer.
|
||||
*/
|
||||
if (!wsTask) {
|
||||
runtimeLog.error(
|
||||
`Auto-merge: ${taskId} workspace partial land but getTask failed (DB outage?) — failing closed, NOT scheduling a retry storm: ${errorMsg}`,
|
||||
);
|
||||
await store
|
||||
.logEntry(
|
||||
taskId,
|
||||
`Workspace partial land — task state unreadable (DB error); parking as failed instead of scheduling a retry storm: ${errorMsg}`,
|
||||
"WorkspacePartialLand",
|
||||
)
|
||||
.catch(() => undefined);
|
||||
await store
|
||||
.updateTask(taskId, { status: "failed", error: errorMsg })
|
||||
.catch(() => undefined);
|
||||
continue;
|
||||
}
|
||||
const wsRetries = wsTask.mergeRetries ?? 0;
|
||||
const decision = shouldRetryAutoMergeConflict(
|
||||
wsRetries,
|
||||
wsSettings as { autoResolveConflicts?: boolean; maxAutoMergeRetries?: unknown } | null,
|
||||
{ skipAutoResolveCheck: true },
|
||||
);
|
||||
await store
|
||||
.logEntry(taskId, `Workspace partial land: ${errorMsg}`, "WorkspacePartialLand")
|
||||
.catch(() => undefined);
|
||||
if (decision.shouldRetry) {
|
||||
await store.updateTask(taskId, { mergeRetries: decision.nextRetryCount, status: null }).catch(() => undefined);
|
||||
const delayMs = 5000 * Math.pow(2, wsRetries);
|
||||
// Capped exponential backoff (B5): cap at 60s so a tuned maxAutoMergeRetries doesn't
|
||||
// push the delay toward ~85 minutes at the ceiling.
|
||||
const delayMs = Math.min(5000 * Math.pow(2, wsRetries), 60_000);
|
||||
runtimeLog.log(
|
||||
`Workspace partial-land retry ${decision.nextRetryCount}/${decision.maxAutoMergeRetries} for ${taskId} in ${delayMs / 1000}s (re-runs skipping landed repos)`,
|
||||
);
|
||||
|
||||
Reference in New Issue
Block a user