feat(workspace): Phase C — per-repo merge loop, land-as-you-go on local integration refs (U5/U6/U7) (#1717)

> ⚠️ **Draft — do not merge until the stack lands.** Stacks on
foundation #1710 + U0 #1711 + Phase A #1713 + Phase B #1714; targets
`main` with the whole stack diff. **Review only the Phase-C commits**
(`744ed09` U1, `7544346` U2, `64e87f9` U3, `627bdcf` review fixes).

## Workspace mode — Phase C (per-repo merge loop, land-as-you-go on
local integration refs)

Implements Phase C of the [master
plan](docs/plans/2026-06-21-002-feat-workspace-mode-execution-model-plan.md)
— units U5/U6/U7 ([Phase-C
plan](docs/plans/2026-06-21-006-feat-workspace-phase-c-plan.md)).
Replaces U0's R7 guard (which *threw* on workspace-task merges) with the
real **per-repo land loop**. A workspace task now runs → captures →
reviews → **merges**: each acquired sub-repo's `fusion/<id>` branch
lands onto **that repo's own LOCAL integration ref** (CAS +
fast-forward, **no remote push** — D2/D5), land-as-you-go.

### What changed
- **U1 — `landOneRepo` + `landWorkspaceTask`.** Extracted the per-repo
land mechanics out of `runAiMerge`'s inline clean-room closure into an
exported `landOneRepo`; `runAiMerge` is now its byte-for-byte
single-repo caller (the 56-test merger-ai oracle stays green).
`landWorkspaceTask` loops the acquired sub-repos (sorted), re-resolves
each repo's integration branch (override-stripped → own `origin/HEAD`),
lands each, aggregates repo-tagged results. The engine dispatch +
user-facing CLI `fn task merge`/dashboard merge doors route workspace
tasks here; `store.mergeTask`/`aiMergeTask`/the `runAiMerge` chokepoint
stay throwing (defense-in-depth).
- **U2 — landed predicate + finalize-once + auto-retry/park.** Each
landed repo's tip is persisted as `workspaceWorktrees[repo].landedSha`;
`isRepoLanded` skips it on retry (idempotent). The task finalizes to
done exactly once after *all* repos land. A partial land raises
`WorkspacePartialLandError` → the engine consumes a `mergeRetry` and
re-runs (skipping landed repos) up to MAX, then operator-parks.
- **U3 — per-repo land lease.** A new `activeSessionRegistry`
`workspace-repo-land` kind serializes concurrent same-sub-repo lands.

### Review (5 personas)
**No P0.** Adversarial verified the two crown-jewel invariants clean:
**no remote push** anywhere in the land loop (CAS + FF-only), and
**exactly one mergeRetry per attempt** with the hard-fail guard unable
to enter the retry loop. Fixed in-branch (`627bdcf`): a **double-land**
bug (a swallowed `landedSha` write could produce a second squash commit
— now propagated, with a `Fusion-Task-Id`-trailer landed-fallback); a
**cross-phase lease clobber** (a merging task could overrun an executing
task's acquire lease — now `taskId`-aware); a **retry storm** under DB
outage (now fails closed); a **status `'merging'` leak**; a
**reachability-gate poison** that demoted fully-merged workspace tasks
(the fast-path now skips them — they're verified by per-repo
`landedSha`); the **dashboard `merged:false`** contradiction;
busy-contention no longer burns the retry quota; backoff capped;
`WorkspacePartialLandError` promoted to a real class.

### Deferred to Phase D
- Self-healing reconcilers for **partial-landed / stuck** workspace
merges (the landed predicate `isRepoLanded` is exported for this).
- The e2e workspace harness.
- Per-repo worktree teardown; extracting a `workspace-merger.ts` module
(`merger-ai.ts` is large); per-sub-repo cwd reachability verification.

### Verification
Gate green: lint, typecheck (29 projects), build, `test:gate` (649+58);
workspace-merger + merger-ai oracle + project-engine **174**.

🤖 Generated with [Claude Code](https://claude.com/claude-code)


<!-- stage-review-badge-begin -->

---

<a href="https://stagereview.app/Runfusion/Fusion/pull/1717">
  <picture>
<source media="(prefers-color-scheme: dark)"
srcset="https://stagereview.app/assets/gh-open-in-stage-dark.svg">
<img src="https://stagereview.app/assets/gh-open-in-stage-light.svg"
alt="Open in Stage">
  </picture>
</a>

<!-- stage-review-badge-end -->

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **New Features**
* Workspace merges now process repositories independently, tracking
per-repo completion for idempotent retries.
* Concurrent land operations on the same sub-repository are serialized
to prevent conflicts.
* Dashboard and CLI now report full merge status for workspace tasks,
including per-repository outcomes.

* **Bug Fixes**
* Improved handling of partial workspace land failures with proper
backoff and retry logic.
  * Stricter validation of workspace configuration repository arrays.

* **Documentation**
  * Added comprehensive Phase C plan for workspace merge behavior.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->
This commit is contained in:
gsxdsm
2026-06-23 20:34:17 -07:00
committed by GitHub
21 changed files with 2860 additions and 228 deletions

View File

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

View File

@@ -0,0 +1,5 @@
---
"@runfusion/fusion": patch
---
Address Phase C workspace merge-loop review feedback. A sub-repo recognized as already-landed via the `Fusion-Task-Id` trailer fallback (when its `landedSha` persist was lost) now resolves and re-records a concrete `landedSha`, so finalize no longer drops it and mis-reports a fully-landed workspace task as a no-op (`mergeConfirmed:false`). A manual merge that hits sub-repo land-lease contention now surfaces the busy error to the user without consuming the persisted `mergeRetries` quota (matching the auto path's separate busy counter). The partial-land retry persists the incremented retry count before arming the backoff timer — a failed write now fails closed instead of looping without consuming budget — and clears the stale busy-contention counter when a real partial land supersedes transient busy failures. The CLI and dashboard merge doors use the shared `isWorkspaceTask` predicate instead of re-inlining the workspace check, and integration-branch shell interpolation in base-commit capture uses POSIX single-quote escaping.

View File

@@ -0,0 +1,5 @@
---
"@runfusion/fusion": minor
---
Workspace mode (Phase C U3): serialize concurrent same-sub-repo lands with a per-repo file-scope lease. When two workspace tasks try to land onto the SAME sub-repo's local integration ref at the same time, the merge phase now registers the sub-repo's absolute path in the path-keyed active-session registry under a distinct `workspace-repo-land` kind before each land and releases it in a `finally` (on land success or failure — no stuck lock). A second task contending for the same sub-repo fast-fails with a retryable `WorkspaceRepoLandBusyError`, which the existing partial-land auto-retry-then-park dispatch handles (consume a `mergeRetry`, re-enqueue with backoff, then operator-park). Disjoint sub-repos lease different paths and never serialize against each other. The lease prevents clean-room ai-merge worktree collisions; ref correctness is already guaranteed by `advanceIntegrationBranchRef`'s CAS (concurrent-advance → rebuild).

View File

@@ -0,0 +1,12 @@
---
"@runfusion/fusion": minor
---
Workspace mode Phase C (U1): per-repo merge loop. Extract `landOneRepo` from the
`runAiMerge` clean-room land closure (single-repo behavior unchanged) and add
`landWorkspaceTask`, which lands each acquired sub-repo's `fusion/<id>` branch onto
that repo's OWN local integration ref (re-resolved per repo with overrides stripped),
land-as-you-go with no remote push. The engine merge dispatch and the user-facing
CLI/dashboard merge doors now route workspace tasks through this loop instead of
throwing; `store.mergeTask`, `aiMergeTask`, and the `runAiMerge` chokepoint keep
throwing `WorkspaceTaskMergeError` as defense-in-depth.

View File

@@ -0,0 +1,15 @@
---
"@runfusion/fusion": minor
---
Workspace mode Phase C (U2): per-repo landed predicate, finalize-once, and idempotent
auto-retry-then-park. `landWorkspaceTask` now records each sub-repo's `landedSha` after
its branch advances that repo's local integration ref, and on a re-run SKIPS any repo
whose recorded `landedSha` is an ancestor of (or equals) its current integration tip — so
an interrupted multi-repo land retries only the un-landed repos and never re-advances an
already-landed ref. When every acquired repo's landed predicate holds, the task moves to
`done` EXACTLY ONCE via the task-global finalize path with an aggregate `mergeDetails`
(representative `commitSha` + a `workspaceLandedShas` map). A partial land (some repos
unlanded) does not move the task done; the engine merge dispatch surfaces it as a
retryable failure that consumes a `mergeRetry` and auto-retries the merge (skipping landed
repos) up to the configured max, then operator-parks the task as failed.

View File

@@ -0,0 +1,151 @@
---
title: "feat: Workspace mode Phase C — per-repo merge loop (land-as-you-go on local integration refs)"
status: active
date: 2026-06-21
type: feat
origin: docs/plans/2026-06-21-002-feat-workspace-mode-execution-model-plan.md (master plan, Phase C / U5·U6·U7)
depth: deep
---
# feat: Workspace mode Phase C — per-repo merge loop (land-as-you-go on local integration refs)
> **ID namespace:** local `U0·U1·U2·U3` decompose master-plan **U5, U6, U7** (+ a Phase-B-deferred extraction).
> **Anchors are feasibility-pending** — a pre-check runs before implementation (as in Phases A/B). Treat `~:` numbers as approximate until verified.
## Summary
Phase C replaces U0's **R7 guard** — which currently makes every workspace-task merge *throw* `WorkspaceTaskMergeError` — with the real **per-repo merge loop**: for each acquired sub-repo, land that repo's `fusion/<id>` branch onto **that repo's LOCAL integration ref** via a repo-scoped clean-room (the `runAiMerge` mechanism, applied per repo), with no remote push. This is **land-as-you-go** (settled **D2/D5**): repos land independently; a partial land (A lands, B fails) leaves A landed locally and is operator-resettable; an unconditional operator escape hatch always exists.
After Phase C a workspace task can fully run → capture → review → **merge**. **Scope out:** self-healing reconcilers + e2e harness (master U8/U9 = Phase D).
**Stacking:** off Phase B (#1714); PR diff includes the stack; must not merge until it lands.
---
## Problem Frame
`runAiMerge` (merger-ai.ts) lands **one** `task.worktree`'s `fusion/<id>` branch into a single clean-room temp worktree and advances **one** local integration ref via `update-ref` CAS (no push). U0 added the **R7 chokepoint guard** `assertNotWorkspaceTaskMerge(task)` so a `workspaceWorktrees`-bearing task fails fast rather than silently mis-merging the single root. Phase C turns that fail-fast into a real loop: iterate the acquired sub-repos, run the clean-room land per repo against that repo's own local integration ref, track which repos have landed (idempotent retry), hold a per-repo file-scope lease during each land, and aggregate a per-repo `MergeResult`. The single-repo `runAiMerge` path is untouched.
---
## Key Technical Decisions
> **OPEN FORKS — to be confirmed by the feasibility pre-check + user before implementation.** Marked `‹FORK›`. The settled semantics (D2/D5) bound them, but the code shape is to verify.
### KTD0 — Extract `workspace-executor.ts` FIRST (Phase-B-deferred maintainability P1)
Before adding the merge loop, move the workspace branches Phase A/B inlined into `executor.ts` (`captureWorkspaceModifiedFiles`, `reviewWorkspacePerRepo`, the per-repo `verifyWorktreeInvariants` block) into `packages/engine/src/workspace-executor.ts` as module-level functions receiving executor state as args; the `if (this.workspaceConfig)` call sites delegate. Pure move + delegate, no behavior change — its own commit, gate-green, before any Phase-C behavior. This keeps the 16k-line file from absorbing the merge loop too.
### KTD1 — Extract `landOneRepo` from `runAiMerge`, then loop it (master U6; D2/D5) — FORK-A RESOLVED
**Verified:** `runAiMerge`'s land sequence (mkdtemp clean room → `git worktree add --detach` → `installWorktreeDependencies` → `mergeAndReview` → `landSquash` → the concurrent-advance CAS retry loop → `activeSessionRegistry` register/unregister) is an **un-factored inline closure** at `merger-ai.ts:1064-1216`, bound to one `projectRootDir`/`integrationBranch`/`branch`; `mergeAndReview`/`finalizeMerged` are module-private. The CAS seam `advanceIntegrationBranchRef` already takes `rootDir`/`integrationBranch` explicitly. **No remote push anywhere** — D2/D5 "no push" confirmed.
So U1 **extracts** an exported `landOneRepo(store, repoRootDir, branch, integrationBranch, options)` from that closure (returns a per-repo `LandResult`), leaving `runAiMerge` as the byte-for-byte single-repo caller. `landWorkspaceTask(task)` loops the acquired sub-repos calling `landOneRepo` per repo, aggregating a repo-tagged result. **`landOneRepo` stays in `merger-ai.ts`** (the private helpers live there); only the thin `landWorkspaceTask` orchestrator may sit in a new `workspace-merger.ts`.
**Per-repo integration branch (P1 the plan missed):** `workspaceWorktrees[repo]` does NOT store the integration branch (acquisition computes it then discards). `landOneRepo` must **re-resolve per repo** with the same override-stripping acquisition uses — `resolveIntegrationBranch(repoRoot, { ...settings, integrationBranch: undefined, baseBranch: undefined })` — so each sub-repo lands on its own `origin/HEAD`, not a shared branch.
**Per-sub-repo prune rooting (correctness):** `pruneExistingAiMergeWorktrees`/`cleanupStaleTempMergeWorktrees` sweep by the `fusion-ai-merge-<taskId>-` prefix; N per-repo clean rooms share the taskId. Root each sweep at the **sub-repo** (`resolveAiMergeRoot(subRepoRoot)`) so one repo's prune cannot race another repo's live clean room for the same task.
### KTD2 — Door table: route the engine + CLI/dashboard doors, keep the rest throwing (master U6) — RESOLVED
Six guard sites. Per-door (FN-5893):
1. **`project-engine.ts:~2300` engine dispatch** → route `workspaceWorktrees`-bearing tasks to `landWorkspaceTask`.
2. **`runAiMerge:~979` chokepoint guard** → STAYS as defense-in-depth for direct single-repo callers (workspace tasks enter via `landWorkspaceTask`, not here).
3. **`store.mergeTask:~11159`** (core, cannot import `@fusion/engine`) → STAYS throwing.
4. **CLI `dashboard.ts:~1312` + `task.ts:~861`** → **route workspace tasks through the engine merge (`landWorkspaceTask`)** instead of `store.mergeTask`, so user-triggered `fn task merge` / the dashboard merge button work on workspace tasks **(user decision: manual merge works in Phase C)**.
5. **`aiMergeTask` (merger.ts:~7666, deprecated)** → STAYS throwing.
### KTD3 — `landedSha`-only per repo; `landWorkspaceTask` finalizes once; auto-retry then park (master U5) — FORK-B RESOLVED
**Verified:** `finalizeMerged`/`finalizeTask` are **task-global** — they write one task-level `mergeDetails` and move the WHOLE task to `done` (`merger-ai.ts:1298-1401`). So `landOneRepo` must advance the ref + record `workspaceWorktrees[repo].landedSha` **only** (no task move). `landWorkspaceTask` calls `finalizeTask`/move-done **exactly once** after every acquired repo's landed predicate is true.
**Landed predicate:** a repo is landed iff `entry.branch` tip is an ancestor of (or equals) its local integration ref tip (or the recorded `landedSha` is present); `landWorkspaceTask` **skips landed repos** (idempotent).
**Partial-land (user decision: auto-retry then park):** repo B fails after A landed → task goes to a non-done state with A's `landedSha` persisted; the failure **consumes a `mergeRetry`** and the engine **auto-retries `landWorkspaceTask`** (skipping landed A, re-attempting B) up to the existing `MAX`, then **operator-parks** (D5 escape hatch as terminal). No new partial-landed status type — `landedSha` on the entry is the only state added (`types.ts:~2256`).
### KTD4 — Per-repo land lease via `activeSessionRegistry` new kind (master U7) — FORK-C RESOLVED
**Verified:** there is NO separate engine file-scope lease — `activeSessionRegistry` (path-keyed, `kind` enum) is the only mechanism (`runAiMerge` already registers the clean room under `kind:"ai-merge"`). Add a new `ActiveSessionKind` `"workspace-repo-land"` keyed on the **sub-repo absolute path**; register before `landOneRepo`, unregister in `finally`. **The lease is for serialization / clean-room-collision avoidance, not ref correctness** — `advanceIntegrationBranchRef`'s CAS already makes interleaved `update-ref` safe (concurrent-advance → rebuild). Set test expectations accordingly.
---
## Implementation Units
> **Standing requirements:** `FNXC:Workspace <yyyy-MM-dd-hh:mm>` comments; a `.changeset/*.md` (`@runfusion/fusion: minor`); FN-5048 (real two-repo git fixture via `_workspace-fixture.ts`; assert local-ref advancement with NO push; fake timers; no mock-the-world); FN-5893 surface enumeration; the merge gate. Branch off Phase B (`gsxdsm/workspace-phase-c`).
### U0. Extract `workspace-executor.ts` (no behavior change)
**Goal:** Move Phase A/B workspace helpers out of `executor.ts` into `workspace-executor.ts`; call sites delegate. Pure refactor.
**Requirements:** KTD0.
**Dependencies:** none.
**Files:** `packages/engine/src/executor.ts`, `packages/engine/src/workspace-executor.ts` (new), existing workspace tests (imports may shift).
**Approach:** Move `captureWorkspaceModifiedFiles`, `reviewWorkspacePerRepo`, the per-repo `verifyWorktreeInvariants` body; pass `store`/`captureModifiedFiles`/etc. as args. No logic change.
**Test scenarios:** the existing Phase A/B workspace suites pass unchanged (the move is correct iff they stay green). `Test expectation: behavior-preserving — existing suites are the oracle.`
**Verification:** All Phase A/B workspace tests + `test:gate` green; `executor.ts` shrinks; no behavior diff.
### U1. Extract `landOneRepo`, loop it in `landWorkspaceTask`, route the doors (master U6)
**Goal:** Land each acquired sub-repo's branch onto its own local integration ref (land-as-you-go, no push), via an extracted `landOneRepo`; route the engine + CLI/dashboard doors.
**Requirements:** KTD1, KTD2.
**Dependencies:** U0.
**Files:** `packages/engine/src/merger-ai.ts` (extract `landOneRepo` from the `:1064-1216` closure; add `landWorkspaceTask`), `packages/engine/src/project-engine.ts` (`~:2300` dispatch → `landWorkspaceTask`), `packages/cli/src/commands/dashboard.ts` (`~:1312`) + `packages/cli/src/commands/task.ts` (`~:861`) (route workspace tasks to the engine merge), optional `packages/engine/src/workspace-merger.ts` (thin orchestrator), `packages/engine/src/__tests__/workspace-merger.test.ts` (new).
**Approach:** Per KTD1/KTD2. **(a)** Extract `landOneRepo(store, repoRootDir, branch, integrationBranch, options)` from the inline closure — `runAiMerge` becomes its single-repo caller, byte-for-byte. **(b)** `landWorkspaceTask` loops the acquired sub-repos: re-resolve each repo's integration branch (override-stripped), root the prune at the sub-repo, call `landOneRepo`, aggregate repo-tagged results. **(c)** Route the engine dispatch + both CLI doors to `landWorkspaceTask` for `workspaceWorktrees`-bearing tasks; `store.mergeTask`/`aiMergeTask`/the `runAiMerge` chokepoint keep throwing (defense-in-depth).
**Execution note:** Real two-repo fixture; commit on each `fusion/<id>`; assert each repo's **local** integration ref advanced and **no remote ref/push** occurred; assert per-sub-repo prune rooting.
**Test scenarios:**
- Two acquired repos, both clean → both local integration refs advance against each repo's own resolved branch; no push/remote ref; result tags both. (happy)
- Repos with different integration branches → each lands on its own (override-stripping works; not a shared branch). (per-repo resolution)
- A conflict in repo B → repo A lands (its `landedSha` recorded); B's result reports the conflict; the task is NOT moved done. (partial — D2/D5)
- The single-repo (non-workspace) `runAiMerge` path → byte-for-byte unchanged (it calls the extracted `landOneRepo`). (regression)
- `store.mergeTask`/`aiMergeTask` with a workspace task → still throws `WorkspaceTaskMergeError`. (defense-in-depth)
- A workspace task via the CLI/dashboard merge door → routes to `landWorkspaceTask` (does not throw). (user-facing door)
**Verification:** Workspace merges land per repo on local refs (no push) via `landOneRepo`; single-repo unchanged; user doors route; non-routed doors stay guarded.
### U2. Per-repo landed predicate + idempotent retry (master U5)
**Goal:** Track landed repos; retry skips them.
**Requirements:** KTD3.
**Dependencies:** U1.
**Files:** `packages/core/src/types.ts` (`workspaceWorktrees[repo].landedSha?`), the loop in U1, `packages/engine/src/__tests__/workspace-merger-idempotency.test.ts` (new).
**Approach:** Per KTD3. `landOneRepo` records `workspaceWorktrees[repo].landedSha` only (no task move); `landWorkspaceTask` calls `finalizeTask`/move-done exactly once after every acquired repo's landed predicate holds. Landed predicate = ancestor check (or `landedSha` present); skip landed repos. Partial-land → non-done state with `landedSha` persisted; the failure **consumes a `mergeRetry`** and is **auto-retried up to `MAX`, then operator-parked** (user decision).
**Test scenarios:**
- Re-running `landWorkspaceTask` after repo A landed + repo B failed → A is skipped (not re-landed), B is retried; A's ref does not move twice. (idempotency — partial land)
- Landed predicate true when branch tip is an ancestor of the integration tip. (predicate)
- `finalizeTask` runs exactly once, only after ALL repos landed (not per-repo). (completion — no premature done)
- Partial-land failure consumes one `mergeRetry`; after `MAX` retries the task is operator-parked, not silently failed. (retry/park)
**Verification:** Partial lands are idempotent on retry; the task moves done exactly once; auto-retry then park works; no double-land.
### U3. Per-repo file-scope lease during land (master U7)
**Goal:** Serialize concurrent same-sub-repo lands.
**Requirements:** KTD4.
**Dependencies:** U1.
**Files:** the lease seam (FORK-C), the loop in U1, `packages/engine/src/__tests__/workspace-merger-lease.test.ts` (new).
**Approach:** Per KTD4. Acquire a per-repo integration-ref lease before each `landOneRepo`, release in `finally`.
**Test scenarios:**
- Two workspace tasks landing the same sub-repo concurrently → serialized (one waits/fails-fast, no interleaved `update-ref`). (concurrency)
- Disjoint sub-repos → land in parallel without contention. (no false serialization)
- Lease released on land failure (no stuck lock). (cleanup)
**Verification:** Same-sub-repo lands serialize; the lease never leaks.
---
## Scope Boundaries
**In scope:** the extraction (U0), the per-repo merge loop + R7-throw replacement (U1), landed predicate + idempotent retry (U2), per-repo lease (U3).
### Deferred to Follow-Up Work (Phase D / master U8·U9)
- Self-healing reconcilers for partial-landed / stuck workspace merges.
- The e2e workspace harness.
- Per-repo worktree teardown (carried residual).
- Remote push of integration refs (explicitly out — D2/D5 are local-ref only).
- Store-level atomic per-repo `workspaceWorktrees` merge (carried residual).
---
## Risks & Dependencies
- **R1 — R7 throw replacement must not weaken the single-repo guard.** Mitigation: KTD2 dispatches only when `workspaceWorktrees` non-empty; untaught doors keep the throw; regression + defense-in-depth tests.
- **R2 — Partial-land leaves inconsistent local state.** Accepted (D2/D5: local + operator-resettable). Mitigation: KTD3 idempotent retry + persisted `landedSha`; the local-ref-only design means no remote pollution.
- **R3 — Clean-room helper reuse across the loop.** `runAiMerge`'s temp-worktree/CAS seams must be callable per repo without cross-repo state bleed. Mitigation: feasibility pre-check verifies the seams; U1 asserts no cross-repo bleed.
- **R4 — Lease vs acquisition-exclusivity confusion.** The Phase-A/U2 acquisition lock and the Phase-C land lease are different scopes. Mitigation: KTD4 distinct kind; test both.
- **R5 — `executor.ts` extraction regression (U0).** Mitigation: behavior-preserving; existing suites are the oracle; gate-green before U1.
- **Stacking dependency:** off Phase B (#1714); diff includes the stack.
---
## Sources & Research
- Master plan (U5/U6/U7, KTD2/KTD4/KTD7, D2/D5, R7).
- This session: `runAiMerge` advances the LOCAL integration ref via `update-ref` CAS (~merger-ai.ts:817/847), no push; the R7 chokepoint guard `assertNotWorkspaceTaskMerge` (~:979) + the door guards; `store.mergeTask` (third path); `SelfHealingManager.cleanupStaleTempMergeWorktrees` prefix sweep.
- Phase A/B (#1713/#1714): per-repo `baseCommitSha`, `activeWorktrees` Set, `workspace-paths.ts`, `_workspace-fixture.ts`, the workspace helpers U0 extracts.

View File

@@ -9,7 +9,6 @@ import {
CentralCore,
AgentStore,
PluginLoader,
assertNotWorkspaceTaskMerge,
getTaskMergeBlocker,
getEnabledPiExtensionPaths,
isEphemeralAgent,
@@ -18,6 +17,7 @@ import {
resolveGlobalDir,
DEFAULT_AGENT_HEARTBEAT_INTERVAL_MS,
isWorkflowColumnsEnabled,
isWorkspaceTask,
resolveColumnFlags,
BUILTIN_CODING_WORKFLOW_IR,
mergeBuiltInZaiProviderModels,
@@ -43,6 +43,7 @@ import {
} from "@fusion/dashboard";
import {
runAiMerge,
landWorkspaceTask,
MissionAutopilot,
MissionExecutionLoop,
HeartbeatMonitor,
@@ -1305,11 +1306,39 @@ export async function runDashboard(port: number, opts: { paused?: boolean; dev?:
// aiMergeTask is soft-deprecated.
//
const onMergeImpl = async (taskId: string) => {
// FNXC:Workspace 2026-06-21-19:05: R7 merge-boundary guard (master-plan U0).
// Reject workspace-mode tasks before any merge work; per-repo merge lands in
// master-plan U6, which removes this guard.
// FNXC:Workspace 2026-06-21-23:40 (Phase C U1, KTD2):
// Dashboard merge button (UI-only mode). A workspace-mode task routes through
// the ENGINE per-repo merge loop `landWorkspaceTask` (each sub-repo lands on its
// own LOCAL integration ref, no push) instead of throwing — manual merge works in
// Phase C (user decision). U0's R7 throw is replaced here by routing; the engine
// chokepoint + store.mergeTask/aiMergeTask keep throwing as defense-in-depth.
const mergeTask = await store.getTask(taskId).catch(() => null);
if (mergeTask) assertNotWorkspaceTaskMerge(mergeTask);
// FNXC:Workspace 2026-06-22-09:30 (Phase C review B10): use the exported `isWorkspaceTask`
// (the engine/CLI canonical predicate) instead of re-inlining the workspaceWorktrees check.
const isWorkspaceMerge = !!mergeTask && isWorkspaceTask(mergeTask);
if (isWorkspaceMerge) {
const workspaceResult = await landWorkspaceTask(store, mergeTask!, cwd, {
agentStore,
});
const latest = await store.getTask(taskId).catch(() => mergeTask!);
// 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: 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",
};
}
const settings = await store.getSettings();
if (getMergeStrategy(settings) === "pull-request") {

View File

@@ -1,5 +1,5 @@
import { TaskStore, COLUMNS, COLUMN_LABELS, CentralCore, assertNotWorkspaceTaskMerge, buildAutoPauseClearPatch, buildManualRetryResetPatch, extractIntentSignature, findNearDuplicates, getTaskDuplicateLineage, reconcileDeterministicDuplicate, runDeterministicDuplicateGuard, type Settings, type Column, type ColumnId, type StepStatus, type AgentLogType, type AgentLogEntry, type IntentSignature, type NearDuplicateCandidate, type NearDuplicateMatch, type TaskDependencyMutation } from "@fusion/core";
import { runAiMerge } from "@fusion/engine";
import { TaskStore, COLUMNS, COLUMN_LABELS, CentralCore, buildAutoPauseClearPatch, buildManualRetryResetPatch, extractIntentSignature, findNearDuplicates, getTaskDuplicateLineage, isWorkspaceTask, reconcileDeterministicDuplicate, runDeterministicDuplicateGuard, type Settings, type Column, type ColumnId, type StepStatus, type AgentLogType, type AgentLogEntry, type IntentSignature, type NearDuplicateCandidate, type NearDuplicateMatch, type TaskDependencyMutation } from "@fusion/core";
import { runAiMerge, landWorkspaceTask } from "@fusion/engine";
import { createInterface } from "node:readline/promises";
import type { PlanningQuestion, PlanningSummary } from "@fusion/core";
import { createSession, submitResponse, RateLimitError, SessionNotFoundError, InvalidSessionStateError } from "@fusion/dashboard/planning";
@@ -851,14 +851,38 @@ export async function runTaskMerge(id: string, projectName?: string) {
console.log(`\n Merging ${id} with AI...\n`);
try {
// FNXC:Workspace 2026-06-21-19:05: R7 merge-boundary guard (master-plan U0).
// Reject workspace-mode tasks before any merge work; per-repo merge lands in
// master-plan U6, which removes this guard.
// FNXC:MergerUnification 2026-06-21-19:05: unified onto runAiMerge (U0).
// The guard lives INSIDE this try so its throw renders via the formatted
// ` ✗ ...` output below instead of the generic top-level bin.ts handler.
// FNXC:Workspace 2026-06-21-23:40 (Phase C U1, KTD2):
// User-triggered `fn task merge`. A workspace-mode task routes through the
// ENGINE per-repo merge loop `landWorkspaceTask` (each sub-repo lands on its own
// LOCAL integration ref, no push) instead of throwing — manual merge works in
// Phase C (user decision). U0's R7 throw is replaced here by routing; the
// engine chokepoint + store.mergeTask/aiMergeTask keep throwing.
const mergeTaskRecord = await store.getTask(id).catch(() => null);
if (mergeTaskRecord) assertNotWorkspaceTaskMerge(mergeTaskRecord);
// FNXC:Workspace 2026-06-22-09:30 (Phase C review B10): use the exported `isWorkspaceTask`
// (the engine/CLI canonical predicate) instead of re-inlining the workspaceWorktrees check.
const isWorkspaceMerge = !!mergeTaskRecord && isWorkspaceTask(mergeTaskRecord);
if (isWorkspaceMerge) {
const workspaceResult = await landWorkspaceTask(store, mergeTaskRecord!, projectPath, {
onAgentText: (delta) => process.stdout.write(delta),
});
console.log();
for (const repo of workspaceResult.repos) {
const label =
repo.status === "landed" ? `landed ${repo.landedSha?.slice(0, 8) ?? ""} → ${repo.integrationBranch}`
: repo.status === "empty" ? "no net changes"
: `failed: ${repo.error ?? "unknown"}`;
console.log(` ${repo.status === "failed" ? "✗" : "✓"} ${repo.repo}: ${label}`);
}
// 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;
}
const result = await runAiMerge(store, projectPath, id, {
onAgentText: (delta) => process.stdout.write(delta),

View File

@@ -167,11 +167,15 @@ export async function loadWorkspaceConfig(rootDir: string): Promise<WorkspaceCon
try {
const raw = await readFile(configPath, "utf-8");
const parsed = JSON.parse(raw) as unknown;
// FNXC:Workspace 2026-06-22-09:30 (Phase C review nit): validate that `repos` is an array
// OF STRINGS, not merely an array. A malformed config (`{ repos: [123, null] }`) would
// otherwise pass and feed non-string values into path joins downstream.
if (
parsed !== null &&
typeof parsed === "object" &&
"repos" in parsed &&
Array.isArray((parsed as { repos: unknown }).repos)
Array.isArray((parsed as { repos: unknown }).repos) &&
(parsed as { repos: unknown[] }).repos.every((r) => typeof r === "string")
) {
const rawRepos = (parsed as { repos: unknown[] }).repos;
const repos = rawRepos.filter((entry): entry is string => isInRootRelativePath(entry, pathMod));

View File

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

View File

@@ -1855,6 +1855,17 @@ export interface MergeDetails {
* `task.mergeRetries`, which counts in-cycle aiMergeTask retries.
*/
transientRecoveryCount?: number;
/**
* FNXC:Workspace 2026-06-22-00:30 (Phase C U2, KTD3):
* Workspace-mode aggregate landed map: sub-repo relative path → the squash sha
* that landed on that repo's local integration ref. Set ONLY by
* `landWorkspaceTask`'s finalize-once after EVERY acquired repo's landed
* predicate holds; the task-level `commitSha` points at one representative
* landed sha (the first sorted landed repo) so the existing `task:merged`
* consumer (which reads `mergeDetails.commitSha`) is satisfied. Empty/absent
* for single-repo tasks.
*/
workspaceLandedShas?: Record<string, string>;
}
/** Represents an agent's checkout lease on a task. */
@@ -2262,8 +2273,17 @@ export interface Task {
* against that sub-repo's RESOLVED integration branch, local-first. It is the
* per-repo analogue of the single-repo base-commit capture and prevents
* cross-repo files-changed inflation when local integration is ahead of origin.
*
* FNXC:Workspace 2026-06-22-00:30 (Phase C U2, KTD3):
* `landedSha` is the per-repo "this repo's branch has landed on its local
* integration ref" marker, set by `landWorkspaceTask` after a sub-repo's squash
* advances that repo's ref. It is the ONLY partial-land state added (no new
* status type): a re-run's landed predicate skips a repo whose `landedSha` is
* present AND whose recorded value is an ancestor of (or equals) the repo's
* integration tip, so an interrupted multi-repo land retries only the un-landed
* repos and never re-advances an already-landed ref (idempotent retry).
*/
workspaceWorktrees?: Record<string, { worktreePath: string; branch: string; baseCommitSha?: string }>;
workspaceWorktrees?: Record<string, { worktreePath: string; branch: string; baseCommitSha?: string; landedSha?: string }>;
steps: TaskStep[];
currentStep: number;
/**
@@ -2652,14 +2672,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;

View File

@@ -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", () => {

View File

@@ -28,9 +28,18 @@ vi.mock("../merger.js", () => ({
VerificationError: testState.VerificationError,
}));
vi.mock("../merger-ai.js", () => ({
runAiMerge: testState.runAiMerge,
}));
// FNXC:Workspace 2026-06-22-09:30 (Phase C review fix): the dispatch's error handler does
// `err instanceof WorkspaceRepoLandBusyError` / `WorkspacePartialLandError` on EVERY merge error
// (these classes are imported from ./merger-ai.js). A bare replacement mock left them undefined,
// so `instanceof undefined` threw on every recovery path (24 pre-existing red tests). Re-export the
// REAL error classes via importOriginal so the instanceof guards evaluate; only runAiMerge is faked.
vi.mock("../merger-ai.js", async (importOriginal) => {
const actual = await importOriginal<typeof import("../merger-ai.js")>();
return {
...actual,
runAiMerge: testState.runAiMerge,
};
});
vi.mock("../runtimes/in-process-runtime.js", () => ({
InProcessRuntime: vi.fn().mockImplementation(function () {

View File

@@ -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,250 @@ 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);
/*
FNXC:Workspace 2026-06-22-09:30 (Phase C review B5b — assert the 60s CAP, not just the first retry):
Advancing 60s once only proves the first 5s timer fired; an UNcapped exponential
(5s,10s,20s,40s,80s,160s,…) would still pass that. Capture EVERY scheduled busy backoff delay
across enough cycles to pass the cap point (busyCount=4 → 5000*2^4 = 80_000ms, clamped to 60_000)
and assert no delay exceeds 60_000 AND the cap is actually reached. Each advance fires the pending
timer → re-enqueue → landWorkspaceTask rejects busy again → next backoff is scheduled.
*/
const scheduledBusyDelays: number[] = [];
// `globalThis.setTimeout` is already the fake-timer impl here (vi.useFakeTimers above).
// Wrap it to record the requested delay, then delegate to the SAME fake timer so the
// fake clock still drives the callback — no real-timer leakage.
const fakeSetTimeout = globalThis.setTimeout;
const setTimeoutSpy = vi
.spyOn(globalThis, "setTimeout")
.mockImplementation(((cb: (...a: unknown[]) => void, ms?: number, ...rest: unknown[]) => {
if (typeof ms === "number") scheduledBusyDelays.push(ms);
return (fakeSetTimeout as (...a: unknown[]) => unknown)(cb, ms, ...rest);
}) as typeof setTimeout);
try {
// Drive enough busy cycles to climb past the cap point (busyCount 0..5 = 6 cycles).
for (let i = 0; i < 6; i++) {
await vi.advanceTimersByTimeAsync(60_000);
}
} finally {
setTimeoutSpy.mockRestore();
}
// The exponential climbed (more than one distinct delay) AND every delay is capped at 60s.
expect(scheduledBusyDelays.length).toBeGreaterThanOrEqual(5);
expect(Math.max(...scheduledBusyDelays)).toBe(60_000);
expect(scheduledBusyDelays.every((d) => d <= 60_000)).toBe(true);
// The cap was actually exercised: at least one delay sits at the 60s ceiling.
expect(scheduledBusyDelays).toContain(60_000);
// Each fired backoff re-enqueued the merge (the contention retry loop is live).
expect(enqueueSpy).toHaveBeenCalledWith("FN-WSH");
await engine.stop();
} finally {
vi.useRealTimers();
}
});
});

View File

@@ -0,0 +1,456 @@
/*
FNXC:Workspace 2026-06-22-00:30 (Phase C U2, KTD3):
Per-repo landed-predicate + finalize-once + idempotent-retry tests. They drive the REAL
`landWorkspaceTask` against a REAL two-repo git fixture (createWorkspaceFixture) under a
NON-git workspace root, asserting LOCAL integration-ref shas directly (FN-5048: real git
only where the invariant requires it; the AI merge/review agents are injected so NO real
AI calls happen and the squash is a plain `git merge --squash`). The retry/park decision
is tested via the engine's narrow exported seam `shouldRetryWorkspacePartialLand` with
fake timers — NOT by spinning real engine retries.
Coverage (FN-5893 surfaces):
- idempotency: re-run after repo A landed + repo B failed → A is SKIPPED (its integration
ref does NOT advance a second time — assert the ref sha is unchanged), B is retried.
- predicate: landed predicate true when branch tip is an ancestor of integration tip;
false otherwise (ref rebuilt / no landedSha).
- no premature done: finalizeTask/move-done runs EXACTLY ONCE, only after BOTH repos land
— assert the task is NOT moved done after the first repo (partial run).
- completion: all repos landed → task reaches done with aggregate mergeDetails
(workspaceLandedShas map + representative commitSha).
- retry/park: a partial-land failure consumes one mergeRetry; after MAX it parks
(shouldRetryWorkspacePartialLand boundary, fake timers).
*/
import { afterEach, describe, expect, it, vi } from "vitest";
import { EventEmitter } from "node:events";
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, 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;
const TASK_ID = "FN-2002";
const BRANCH = "fusion/fn-2002";
function configureIdentity(dir: string): void {
execSync('git config user.email "test@example.com"', { cwd: dir, stdio: "pipe" });
execSync('git config user.name "Test"', { cwd: dir, stdio: "pipe" });
}
interface RecordingStore extends EventEmitter {
task: Task;
moveTaskCalls: Array<{ id: string; column: string }>;
emitted: Array<{ event: string; payload: unknown }>;
}
/**
* A store that PERSISTS workspaceWorktrees + mergeDetails updates on a single in-memory
* task and returns it from getTask, so the landed-predicate retry reads back the
* `landedSha` that landWorkspaceTask wrote (real fresh-read-then-merge behavior).
*/
function createStore(task: Task, settings: Record<string, unknown> = {}): TaskStore & RecordingStore {
const emitter = new EventEmitter();
const moveTaskCalls: Array<{ id: string; column: string }> = [];
const emitted: Array<{ event: string; payload: unknown }> = [];
const realEmit = emitter.emit.bind(emitter);
const store = Object.assign(emitter, {
task,
moveTaskCalls,
emitted,
getSettings: vi.fn().mockResolvedValue({ autoMerge: false, ...settings }),
updateTask: vi.fn(async (_id: string, patch: Partial<Task>) => {
Object.assign(store.task, patch);
return undefined;
}),
logEntry: vi.fn().mockResolvedValue(undefined),
appendAgentLog: vi.fn().mockResolvedValue(undefined),
getTask: vi.fn(async () => store.task),
moveTask: vi.fn((id: string, column: string) => {
moveTaskCalls.push({ id, column });
store.task.column = column as Task["column"];
return Promise.resolve(store.task);
}),
upsertTaskCommitAssociation: vi.fn().mockResolvedValue(undefined),
accumulateTokenUsage: vi.fn().mockResolvedValue(undefined),
emit: (event: string, payload?: unknown) => {
emitted.push({ event, payload });
return realEmit(event, payload);
},
}) as unknown as TaskStore & RecordingStore;
return store;
}
/** Add a real `fusion/<id>` branch to a sub-repo with one own non-conflicting commit. */
function addRepoBranchWithEdit(fx: WorkspaceFixture, repoRel: string, content: string): void {
const repoDir = fx.repoPath(repoRel);
const worktreePath = path.join(repoDir, ".wt-branch");
fx.git(repoRel, `git worktree add -b ${BRANCH} ${worktreePath} HEAD`);
configureIdentity(worktreePath);
writeFileSync(path.join(worktreePath, "feature.txt"), content, "utf-8");
execSync("git add feature.txt", { cwd: worktreePath, stdio: "pipe" });
execSync(`git commit -m "feat(${TASK_ID}): add feature in ${repoRel}"`, { cwd: worktreePath, stdio: "pipe" });
fx.git(repoRel, `git worktree remove --force ${worktreePath}`);
}
/** Make a sub-repo's integration tip + task branch BOTH edit README → squash conflicts. */
function makeConflictingRepo(fx: WorkspaceFixture, repoRel: string): void {
const repoDir = fx.repoPath(repoRel);
const worktreePath = path.join(repoDir, ".wt-conflict");
fx.git(repoRel, `git worktree add -b ${BRANCH} ${worktreePath} HEAD`);
configureIdentity(worktreePath);
writeFileSync(path.join(worktreePath, "README.md"), "# branch-side change\n", "utf-8");
execSync("git add README.md", { cwd: worktreePath, stdio: "pipe" });
execSync(`git commit -m "feat(${TASK_ID}): branch README"`, { cwd: worktreePath, stdio: "pipe" });
fx.git(repoRel, `git worktree remove --force ${worktreePath}`);
writeFileSync(path.join(repoDir, "README.md"), "# main-side change\n", "utf-8");
fx.git(repoRel, "git add README.md");
fx.git(repoRel, 'git commit -m "main diverge README"');
}
/** Resolve repo-b's conflict by replacing the conflicting README content (no markers). */
function resolveConflictInRepo(fx: WorkspaceFixture, repoRel: string): void {
// Re-point the task branch so the squash no longer conflicts: drop the branch's
// README edit and add a clean feature file instead.
const repoDir = fx.repoPath(repoRel);
fx.git(repoRel, `git branch -D ${BRANCH}`);
const worktreePath = path.join(repoDir, ".wt-resolved");
fx.git(repoRel, `git worktree add -b ${BRANCH} ${worktreePath} HEAD`);
configureIdentity(worktreePath);
writeFileSync(path.join(worktreePath, "feature.txt"), "resolved feature\n", "utf-8");
execSync("git add feature.txt", { cwd: worktreePath, stdio: "pipe" });
execSync(`git commit -m "feat(${TASK_ID}): resolved"`, { cwd: worktreePath, stdio: "pipe" });
fx.git(repoRel, `git worktree remove --force ${worktreePath}`);
}
/** A merge agent that performs the real squash in the clean room (no AI). */
function squashMergeAgent(branch: string) {
return async (cwd: string): Promise<void> => {
configureIdentity(cwd);
try {
execSync(`git merge --squash ${branch}`, { cwd, stdio: "pipe" });
} catch {
// squash reported conflicts — fall through to the unmerged check.
}
const unmerged = execSync("git ls-files -u", { cwd, encoding: "utf-8" }).trim();
if (unmerged.length > 0) {
throw new Error("merge conflict: unresolved paths in clean room");
}
const staged = execSync("git diff --cached --name-only", { cwd, encoding: "utf-8" }).trim();
if (staged.length === 0) return;
execSync(`git commit -m "${branch}: squashed"`, { cwd, stdio: "pipe" });
};
}
const approveReviewAgent = async (): Promise<string> => "REVIEW_VERDICT: approve";
function makeTask(workspaceWorktrees: Task["workspaceWorktrees"]): Task {
return {
id: TASK_ID,
title: "Workspace merge task",
description: "",
column: "in-review",
branch: BRANCH,
dependencies: [],
steps: [],
currentStep: 0,
log: [],
workspaceWorktrees,
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
} as Task;
}
describeIfGit("landWorkspaceTask — landed predicate + finalize-once + idempotent retry (Phase C U2)", () => {
let fx: WorkspaceFixture;
afterEach(() => fx?.cleanup());
it("idempotency: re-run after A landed + B failed skips A (ref unchanged) and retries B", async () => {
fx = await createWorkspaceFixture(["repo-a", "repo-b"]);
addRepoBranchWithEdit(fx, "repo-a", "a feature\n");
makeConflictingRepo(fx, "repo-b");
const task = makeTask({
"repo-a": { worktreePath: fx.repoPath("repo-a"), branch: BRANCH },
"repo-b": { worktreePath: fx.repoPath("repo-b"), branch: BRANCH },
});
const store = createStore(task);
// First run: A lands, B conflicts → partial.
const first = await landWorkspaceTask(store, store.task, fx.rootDir, {}, {
mergeAgent: squashMergeAgent(BRANCH),
reviewAgent: approveReviewAgent,
});
expect(first.allLanded).toBe(false);
expect(first.finalized).toBe(false);
const tipAAfterFirst = fx.git("repo-a", "git rev-parse refs/heads/main");
// A's landedSha was persisted onto the task entry.
expect(store.task.workspaceWorktrees!["repo-a"].landedSha).toBe(tipAAfterFirst);
// Not moved done on a partial land.
expect(store.moveTaskCalls).toHaveLength(0);
// Operator resolves repo B's conflict, then the merge is re-run (auto-retry).
resolveConflictInRepo(fx, "repo-b");
const second = await landWorkspaceTask(store, store.task, fx.rootDir, {}, {
mergeAgent: squashMergeAgent(BRANCH),
reviewAgent: approveReviewAgent,
});
// A was SKIPPED (already landed): its integration ref did NOT advance a second time.
expect(fx.git("repo-a", "git rev-parse refs/heads/main")).toBe(tipAAfterFirst);
const repoA = second.repos.find((r) => r.repo === "repo-a")!;
expect(repoA.alreadyLanded).toBe(true);
expect(repoA.status).toBe("landed");
// B was retried and landed this time.
const repoB = second.repos.find((r) => r.repo === "repo-b")!;
expect(repoB.status).toBe("landed");
expect(repoB.alreadyLanded).toBeFalsy();
expect(second.allLanded).toBe(true);
// Finalize-once ran on the completing run.
expect(second.finalized).toBe(true);
expect(store.moveTaskCalls).toEqual([{ id: TASK_ID, column: "done" }]);
});
it("predicate: landedSha that is an ancestor of the integration tip reads as landed; a non-ancestor does not", 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 } });
const store = createStore(task);
// Land repo-a once.
const first = await landWorkspaceTask(store, store.task, fx.rootDir, {}, {
mergeAgent: squashMergeAgent(BRANCH),
reviewAgent: approveReviewAgent,
});
expect(first.allLanded).toBe(true);
const landedSha = store.task.workspaceWorktrees!["repo-a"].landedSha!;
const tip = fx.git("repo-a", "git rev-parse refs/heads/main");
// landedSha == tip → ancestor-or-equal → landed. Advance main with an UNRELATED
// commit; the landedSha is still an ancestor, so it must STILL read as landed.
writeFileSync(path.join(fx.repoPath("repo-a"), "unrelated.txt"), "x\n", "utf-8");
fx.git("repo-a", "git add unrelated.txt");
fx.git("repo-a", 'git commit -m "unrelated advance"');
expect(fx.git("repo-a", "git merge-base --is-ancestor " + landedSha + " refs/heads/main && echo yes").trim()).toBe("yes");
// Re-run: predicate true (ancestor) → repo skipped, no re-land.
const tipBeforeRerun = fx.git("repo-a", "git rev-parse refs/heads/main");
const second = await landWorkspaceTask(store, store.task, fx.rootDir, {}, {
mergeAgent: squashMergeAgent(BRANCH),
reviewAgent: approveReviewAgent,
});
expect(second.repos[0].alreadyLanded).toBe(true);
expect(fx.git("repo-a", "git rev-parse refs/heads/main")).toBe(tipBeforeRerun);
// Non-ancestor: reset main to before the landedSha → landedSha no longer reachable →
// predicate false → the repo re-lands.
void tip;
fx.git("repo-a", "git reset --hard HEAD~2"); // before the squash + unrelated commit
const tipReset = fx.git("repo-a", "git rev-parse refs/heads/main");
expect(fx.git("repo-a", `git merge-base --is-ancestor ${landedSha} refs/heads/main || echo no`).trim()).toBe("no");
const third = await landWorkspaceTask(store, store.task, fx.rootDir, {}, {
mergeAgent: squashMergeAgent(BRANCH),
reviewAgent: approveReviewAgent,
});
expect(third.repos[0].alreadyLanded).toBeFalsy();
expect(third.repos[0].status).toBe("landed");
expect(fx.git("repo-a", "git rev-parse refs/heads/main")).not.toBe(tipReset);
});
it("no premature done: a partial run (one repo failed) does NOT move the task done", async () => {
fx = await createWorkspaceFixture(["repo-a", "repo-b"]);
addRepoBranchWithEdit(fx, "repo-a", "a feature\n");
makeConflictingRepo(fx, "repo-b");
const task = makeTask({
"repo-a": { worktreePath: fx.repoPath("repo-a"), branch: BRANCH },
"repo-b": { worktreePath: fx.repoPath("repo-b"), branch: BRANCH },
});
const store = createStore(task);
const result = await landWorkspaceTask(store, store.task, fx.rootDir, {}, {
mergeAgent: squashMergeAgent(BRANCH),
reviewAgent: approveReviewAgent,
});
// repo-a landed first, but the task must NOT be done because repo-b failed.
expect(result.repos.find((r) => r.repo === "repo-a")!.status).toBe("landed");
expect(result.finalized).toBe(false);
expect(store.moveTaskCalls).toHaveLength(0);
expect(store.emitted.some((e) => e.event === "task:merged")).toBe(false);
});
it("completion: all repos landed → task moves done ONCE with aggregate mergeDetails", async () => {
fx = await createWorkspaceFixture(["repo-a", "repo-b"]);
addRepoBranchWithEdit(fx, "repo-a", "a feature\n");
addRepoBranchWithEdit(fx, "repo-b", "b feature\n");
const task = makeTask({
"repo-a": { worktreePath: fx.repoPath("repo-a"), branch: BRANCH },
"repo-b": { worktreePath: fx.repoPath("repo-b"), branch: BRANCH },
});
const store = createStore(task);
const result = await landWorkspaceTask(store, store.task, fx.rootDir, {}, {
mergeAgent: squashMergeAgent(BRANCH),
reviewAgent: approveReviewAgent,
});
expect(result.allLanded).toBe(true);
expect(result.finalized).toBe(true);
// Moved done exactly once and emitted task:merged exactly once.
expect(store.moveTaskCalls).toEqual([{ id: TASK_ID, column: "done" }]);
const mergedEvents = store.emitted.filter((e) => e.event === "task:merged");
expect(mergedEvents).toHaveLength(1);
// Aggregate mergeDetails: a representative commitSha + the per-repo landed map.
const md = store.task.mergeDetails!;
expect(md.mergeConfirmed).toBe(true);
const landedShaA = fx.git("repo-a", "git rev-parse refs/heads/main");
const landedShaB = fx.git("repo-b", "git rev-parse refs/heads/main");
expect(md.workspaceLandedShas).toEqual({ "repo-a": landedShaA, "repo-b": landedShaB });
// commitSha is one of the landed repo shas (representative for the task:merged consumer).
expect([landedShaA, landedShaB]).toContain(md.commitSha);
});
});
/*
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();
});
});
// FNXC:Workspace 2026-06-22-09:30 (Phase C review nit): the former generic "fake-timer backoff
// schedule does not spin real retries" smoke test only proved Vitest's fake timers work — it never
// drove the production retry seam. The real backoff-cap invariant is now asserted against the live
// ProjectEngine in project-engine.test.ts ("B4/B5: busy contention re-enqueues with capped backoff").
describe("workspace partial-land retry/park decision (engine seam)", () => {
it("consumes a mergeRetry up to MAX, then parks (shouldRetryWorkspacePartialLand)", () => {
// Default MAX = 3. currentRetries + 1 < MAX gates retry.
expect(shouldRetryWorkspacePartialLand(0, {})).toMatchObject({
shouldRetry: true,
maxAutoMergeRetries: 3,
nextRetryCount: 1,
});
expect(shouldRetryWorkspacePartialLand(1, {})).toMatchObject({
shouldRetry: true,
maxAutoMergeRetries: 3,
nextRetryCount: 2,
});
// Last attempt: currentRetries + 1 === MAX → park (no further retry).
expect(shouldRetryWorkspacePartialLand(2, {})).toMatchObject({
shouldRetry: false,
maxAutoMergeRetries: 3,
nextRetryCount: 3,
});
// Custom cap honored.
expect(shouldRetryWorkspacePartialLand(3, { maxAutoMergeRetries: 5 }).shouldRetry).toBe(true);
expect(shouldRetryWorkspacePartialLand(4, { maxAutoMergeRetries: 5 }).shouldRetry).toBe(false);
});
});

View File

@@ -0,0 +1,318 @@
/*
FNXC:Workspace 2026-06-22-02:10 (Phase C U3, KTD4):
Per-repo LAND lease tests. They drive the REAL `landWorkspaceTask` against a REAL
two-repo git fixture (createWorkspaceFixture) and assert the lease seam directly on
the REAL module-level `activeSessionRegistry` singleton (FN-5048: narrow seam — we
assert registry state + a merge-agent spy, NO real concurrent processes, NO
mock-the-world; the AI merge/review agents are injected so no real AI calls happen
and the squash is a plain `git merge --squash`).
The lease is keyed by the sub-repo ABSOLUTE path under kind "workspace-repo-land".
It is for SERIALIZATION / clean-room-collision avoidance only — `advanceIntegration
BranchRef`'s CAS already makes the interleaved `update-ref` correct — so we assert
serialization behavior (one wins, the other fast-fails) and that the lease never leaks.
Coverage (FN-5893 surfaces):
- concurrency: two tasks landing the SAME sub-repo → one acquires the land lease,
the other FAST-FAILS with WorkspaceRepoLandBusyError; no interleaved update-ref on
that repo's ref (the loser advances nothing). Lease kind/path asserted while held.
- independence: disjoint sub-repos (task1→repo-a, task2→repo-b) → both proceed, no
false serialization (neither sees the other's lease path).
- cleanup: a repo land that THROWS → the lease for that path is released (not stuck),
so a subsequent land of the same repo can acquire it.
*/
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { EventEmitter } from "node:events";
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, WorkspaceRepoLandBusyError } from "../merger-ai.js";
import { activeSessionRegistry } from "../active-session-registry.js";
import { createWorkspaceFixture, hasGit, type WorkspaceFixture } from "./_workspace-fixture.js";
const describeIfGit = hasGit ? describe : describe.skip;
const BRANCH = "fusion/fn-3003";
const LAND_KIND = "workspace-repo-land";
function configureIdentity(dir: string): void {
execSync('git config user.email "test@example.com"', { cwd: dir, stdio: "pipe" });
execSync('git config user.name "Test"', { cwd: dir, stdio: "pipe" });
}
interface RecordingStore extends EventEmitter {
task: Task;
moveTaskCalls: Array<{ id: string; column: string }>;
}
/** A store that persists workspaceWorktrees/mergeDetails on one in-memory task. */
function createStore(task: Task): TaskStore & RecordingStore {
const emitter = new EventEmitter();
const moveTaskCalls: Array<{ id: string; column: string }> = [];
const store = Object.assign(emitter, {
task,
moveTaskCalls,
getSettings: vi.fn().mockResolvedValue({ autoMerge: false }),
updateTask: vi.fn(async (_id: string, patch: Partial<Task>) => {
Object.assign(store.task, patch);
return undefined;
}),
logEntry: vi.fn().mockResolvedValue(undefined),
appendAgentLog: vi.fn().mockResolvedValue(undefined),
getTask: vi.fn(async () => store.task),
moveTask: vi.fn((id: string, column: string) => {
moveTaskCalls.push({ id, column });
store.task.column = column as Task["column"];
return Promise.resolve(store.task);
}),
upsertTaskCommitAssociation: vi.fn().mockResolvedValue(undefined),
accumulateTokenUsage: vi.fn().mockResolvedValue(undefined),
}) as unknown as TaskStore & RecordingStore;
return store;
}
/** Add a real `fusion/<id>` branch to a sub-repo with one own non-conflicting commit. */
function addRepoBranchWithEdit(fx: WorkspaceFixture, repoRel: string, taskId: string, content: string): void {
const repoDir = fx.repoPath(repoRel);
const worktreePath = path.join(repoDir, `.wt-${taskId}`);
fx.git(repoRel, `git worktree add -b ${BRANCH} ${worktreePath} HEAD`);
configureIdentity(worktreePath);
writeFileSync(path.join(worktreePath, "feature.txt"), content, "utf-8");
execSync("git add feature.txt", { cwd: worktreePath, stdio: "pipe" });
execSync(`git commit -m "feat(${taskId}): add feature in ${repoRel}"`, { cwd: worktreePath, stdio: "pipe" });
fx.git(repoRel, `git worktree remove --force ${worktreePath}`);
}
/** A merge agent that performs the real squash in the clean room (no AI). */
function squashMergeAgent(branch: string, onEnter?: (cwd: string) => void | Promise<void>) {
return async (cwd: string): Promise<void> => {
if (onEnter) await onEnter(cwd);
configureIdentity(cwd);
try {
execSync(`git merge --squash ${branch}`, { cwd, stdio: "pipe" });
} catch {
// squash reported conflicts — fall through to the unmerged check.
}
const unmerged = execSync("git ls-files -u", { cwd, encoding: "utf-8" }).trim();
if (unmerged.length > 0) throw new Error("merge conflict: unresolved paths in clean room");
const staged = execSync("git diff --cached --name-only", { cwd, encoding: "utf-8" }).trim();
if (staged.length === 0) return;
execSync(`git commit -m "${branch}: squashed"`, { cwd, stdio: "pipe" });
};
}
const approveReviewAgent = async (): Promise<string> => "REVIEW_VERDICT: approve";
function makeTask(id: string, workspaceWorktrees: Task["workspaceWorktrees"]): Task {
return {
id,
title: "Workspace merge task",
description: "",
column: "in-review",
branch: BRANCH,
dependencies: [],
steps: [],
currentStep: 0,
log: [],
workspaceWorktrees,
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
} as Task;
}
describeIfGit("landWorkspaceTask — per-repo land lease (Phase C U3, KTD4)", () => {
let fx: WorkspaceFixture;
afterEach(() => {
fx?.cleanup();
activeSessionRegistry.clear();
vi.restoreAllMocks();
});
beforeEach(() => activeSessionRegistry.clear());
it("concurrency: two tasks landing the SAME sub-repo serialize — one acquires the land lease, the other fast-fails (no interleaved update-ref)", async () => {
fx = await createWorkspaceFixture(["repo-a"]);
addRepoBranchWithEdit(fx, "repo-a", "FN-3001", "a feature\n");
const repoAbs = fx.repoPath("repo-a");
const task1 = makeTask("FN-3001", { "repo-a": { worktreePath: repoAbs, branch: BRANCH } });
const task2 = makeTask("FN-3001", { "repo-a": { worktreePath: repoAbs, branch: BRANCH } });
// Distinct task IDs so the lease owner check (taskId !== holder) triggers.
task2.id = "FN-3002";
const store1 = createStore(task1);
const store2 = createStore(task2);
let loserError: unknown;
const tipBefore = fx.git("repo-a", "git rev-parse refs/heads/main");
// task1's merge agent blocks until task2 has tried (and failed) to acquire the
// land lease for the SAME sub-repo path. While task1 holds the lease we assert it
// is registered under the right kind + path; task2 fast-fails with the busy error.
const winner = landWorkspaceTask(store1, store1.task, fx.rootDir, {}, {
mergeAgent: squashMergeAgent(BRANCH, async () => {
// task1 now holds the land lease for repo-a.
const held = activeSessionRegistry.lookupByPath(repoAbs);
expect(held?.kind).toBe(LAND_KIND);
expect(held?.taskId).toBe("FN-3001");
// task2 attempts the same sub-repo concurrently → must fast-fail.
try {
await landWorkspaceTask(store2, store2.task, fx.rootDir, {}, {
mergeAgent: squashMergeAgent(BRANCH),
reviewAgent: approveReviewAgent,
});
} catch (err) {
loserError = err;
}
// The loser advanced NOTHING: the ref is still at the pre-land tip.
expect(fx.git("repo-a", "git rev-parse refs/heads/main")).toBe(tipBefore);
}),
reviewAgent: approveReviewAgent,
});
const result = await winner;
// Winner landed.
expect(result.allLanded).toBe(true);
expect(result.repos[0].status).toBe("landed");
expect(fx.git("repo-a", "git rev-parse refs/heads/main")).not.toBe(tipBefore);
// Loser fast-failed with the retryable busy error (serialized, not broken).
expect(loserError).toBeInstanceOf(WorkspaceRepoLandBusyError);
expect((loserError as WorkspaceRepoLandBusyError).retryable).toBe(true);
expect((loserError as WorkspaceRepoLandBusyError).holderTaskId).toBe("FN-3001");
// Lease released after the winner finished — no leak.
expect(activeSessionRegistry.lookupByPath(repoAbs)).toBeNull();
});
it("independence: disjoint sub-repos land without contention (no false serialization)", async () => {
fx = await createWorkspaceFixture(["repo-a", "repo-b"]);
addRepoBranchWithEdit(fx, "repo-a", "FN-3001", "a feature\n");
addRepoBranchWithEdit(fx, "repo-b", "FN-3002", "b feature\n");
const repoAAbs = fx.repoPath("repo-a");
const repoBAbs = fx.repoPath("repo-b");
const task1 = makeTask("FN-3001", { "repo-a": { worktreePath: repoAAbs, branch: BRANCH } });
const task2 = makeTask("FN-3002", { "repo-b": { worktreePath: repoBAbs, branch: BRANCH } });
const store1 = createStore(task1);
const store2 = createStore(task2);
let task2Error: unknown;
let task2Landed = false;
// task1 lands repo-a; mid-land it kicks off task2 landing the DISJOINT repo-b.
// task2 leases a DIFFERENT path, so it must NOT serialize against task1.
const t1 = landWorkspaceTask(store1, store1.task, fx.rootDir, {}, {
mergeAgent: squashMergeAgent(BRANCH, async () => {
// While task1 holds repo-a's lease, repo-b's lease is unheld.
expect(activeSessionRegistry.lookupByPath(repoAAbs)?.kind).toBe(LAND_KIND);
expect(activeSessionRegistry.lookupByPath(repoBAbs)).toBeNull();
try {
const r2 = await landWorkspaceTask(store2, store2.task, fx.rootDir, {}, {
mergeAgent: squashMergeAgent(BRANCH),
reviewAgent: approveReviewAgent,
});
task2Landed = r2.allLanded;
} catch (err) {
task2Error = err;
}
}),
reviewAgent: approveReviewAgent,
});
const r1 = await t1;
// Both proceeded — no false serialization.
expect(task2Error).toBeUndefined();
expect(task2Landed).toBe(true);
expect(r1.allLanded).toBe(true);
expect(fx.git("repo-a", "git rev-parse refs/heads/main")).not.toBe(
fx.git("repo-a", "git rev-parse fusion/fn-3003^"),
);
// Both leases released.
expect(activeSessionRegistry.lookupByPath(repoAAbs)).toBeNull();
expect(activeSessionRegistry.lookupByPath(repoBAbs)).toBeNull();
});
it("cleanup: a land failure releases the lease (not stuck) so a subsequent land can acquire", async () => {
fx = await createWorkspaceFixture(["repo-a"]);
addRepoBranchWithEdit(fx, "repo-a", "FN-3001", "a feature\n");
const repoAbs = fx.repoPath("repo-a");
const task = makeTask("FN-3001", { "repo-a": { worktreePath: repoAbs, branch: BRANCH } });
const store = createStore(task);
// A merge agent that throws → landOneRepo fails → the per-repo land lease finally
// must release the lease even on failure.
const throwingAgent = async (): Promise<void> => {
// Lease is held at this point.
expect(activeSessionRegistry.lookupByPath(repoAbs)?.kind).toBe(LAND_KIND);
throw new Error("synthetic clean-room failure");
};
const failed = await landWorkspaceTask(store, store.task, fx.rootDir, {}, {
mergeAgent: throwingAgent,
reviewAgent: approveReviewAgent,
});
expect(failed.allLanded).toBe(false);
expect(failed.repos[0].status).toBe("failed");
// Lease was released despite the failure — NOT stuck.
expect(activeSessionRegistry.lookupByPath(repoAbs)).toBeNull();
// A subsequent land of the SAME repo can acquire (real squash this time).
const retry = await landWorkspaceTask(store, store.task, fx.rootDir, {}, {
mergeAgent: squashMergeAgent(BRANCH),
reviewAgent: approveReviewAgent,
});
expect(retry.allLanded).toBe(true);
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();
});
});

View File

@@ -0,0 +1,315 @@
/*
FNXC:Workspace 2026-06-21-23:40 (Phase C U1, KTD1/KTD2):
Per-repo workspace merge-loop tests. They drive the REAL `landWorkspaceTask` /
`landOneRepo` against a REAL two-repo git fixture under a NON-git workspace root
(createWorkspaceFixture), so a leaked rootDir git preflight would actually fail and a
shared clean-room root would race. Real git is used only where the invariant requires
it (the local-ref advance, the no-push assertion); the AI merge/review agents are
injected (deps) so NO real AI calls happen and the squash is produced by a plain
`git merge --squash` inside the clean room — no mock-the-world child_process.
Coverage (FN-5893 surfaces):
- happy: two acquired repos both clean → BOTH local integration refs advance against
each repo's own resolved branch; NO remote ref/push happened; result tags both. Since
Phase C U2, a fully-landed workspace task also finalizes ONCE (moves done, emits
task:merged) — asserted here; the landed-predicate/finalize-once/retry mechanics have
dedicated coverage in workspace-merger-idempotency.test.ts.
- per-repo resolution: repos with DIFFERENT origin/HEAD integration branches → each
lands on its own (override-stripping works, not a shared branch).
- partial: a conflict in repo B → repo A lands (landedSha recorded); B reports the
failure; the task is NOT moved done (no finalizeTask call) — the partial-land retry is U2.
- defense-in-depth: store.mergeTask / aiMergeTask with a workspace task → still throw
WorkspaceTaskMergeError.
The single-repo runAiMerge regression lives in the existing merger-ai*.test.ts (the
extraction is byte-for-byte; runAiMerge is landOneRepo's single-repo caller).
*/
import { afterEach, describe, expect, it, vi } from "vitest";
import { EventEmitter } from "node:events";
import { execSync } from "node:child_process";
import { writeFileSync } from "node:fs";
import path from "node:path";
import type { Task, TaskStore } from "@fusion/core";
import { assertNotWorkspaceTaskMerge } from "@fusion/core";
import { landWorkspaceTask, runAiMerge } from "../merger-ai.js";
import { createWorkspaceFixture, hasGit, type WorkspaceFixture } from "./_workspace-fixture.js";
const describeIfGit = hasGit ? describe : describe.skip;
const TASK_ID = "FN-2001";
const BRANCH = "fusion/fn-2001";
function configureIdentity(dir: string): void {
execSync('git config user.email "test@example.com"', { cwd: dir, stdio: "pipe" });
execSync('git config user.name "Test"', { cwd: dir, stdio: "pipe" });
}
interface RecordingStore extends EventEmitter {
moveTaskCalls: Array<{ id: string; column: string }>;
emitted: Array<{ event: string; payload: unknown }>;
}
function createStore(settings: Record<string, unknown> = {}): TaskStore & RecordingStore {
const emitter = new EventEmitter();
const moveTaskCalls: Array<{ id: string; column: string }> = [];
const emitted: Array<{ event: string; payload: unknown }> = [];
const realEmit = emitter.emit.bind(emitter);
const store = Object.assign(emitter, {
moveTaskCalls,
emitted,
getSettings: vi.fn().mockResolvedValue({ autoMerge: false, ...settings }),
updateTask: vi.fn().mockResolvedValue(undefined),
logEntry: vi.fn().mockResolvedValue(undefined),
appendAgentLog: vi.fn().mockResolvedValue(undefined),
getTask: vi.fn().mockResolvedValue(undefined),
moveTask: vi.fn((id: string, column: string) => {
moveTaskCalls.push({ id, column });
return Promise.resolve({ id, column } as Task);
}),
upsertTaskCommitAssociation: vi.fn().mockResolvedValue(undefined),
accumulateTokenUsage: vi.fn().mockResolvedValue(undefined),
emit: (event: string, payload?: unknown) => {
emitted.push({ event, payload });
return realEmit(event, payload);
},
}) as unknown as TaskStore & RecordingStore;
return store;
}
/**
* Add a real `fusion/<id>` worktree to a sub-repo with one own commit that EDITS the
* README the integration tip already has, then remove the worktree (we only need the
* branch ref). Returns the branch name. By default the edit is non-conflicting.
*/
function addRepoBranchWithEdit(fx: WorkspaceFixture, repoRel: string, content: string): void {
const repoDir = fx.repoPath(repoRel);
const worktreePath = path.join(repoDir, ".wt-branch");
fx.git(repoRel, `git worktree add -b ${BRANCH} ${worktreePath} HEAD`);
configureIdentity(worktreePath);
writeFileSync(path.join(worktreePath, "feature.txt"), content, "utf-8");
execSync("git add feature.txt", { cwd: worktreePath, stdio: "pipe" });
execSync(`git commit -m "feat(${TASK_ID}): add feature in ${repoRel}"`, { cwd: worktreePath, stdio: "pipe" });
fx.git(repoRel, `git worktree remove --force ${worktreePath}`);
}
/** Make a sub-repo's integration tip and the task branch BOTH edit README so the
* squash conflicts. */
function makeConflictingRepo(fx: WorkspaceFixture, repoRel: string): void {
const repoDir = fx.repoPath(repoRel);
// Task branch edits README on a new commit.
const worktreePath = path.join(repoDir, ".wt-conflict");
fx.git(repoRel, `git worktree add -b ${BRANCH} ${worktreePath} HEAD`);
configureIdentity(worktreePath);
writeFileSync(path.join(worktreePath, "README.md"), "# branch-side change\n", "utf-8");
execSync("git add README.md", { cwd: worktreePath, stdio: "pipe" });
execSync(`git commit -m "feat(${TASK_ID}): branch README"`, { cwd: worktreePath, stdio: "pipe" });
fx.git(repoRel, `git worktree remove --force ${worktreePath}`);
// Integration tip (main) diverges with a conflicting README edit.
writeFileSync(path.join(repoDir, "README.md"), "# main-side change\n", "utf-8");
fx.git(repoRel, "git add README.md");
fx.git(repoRel, 'git commit -m "main diverge README"');
}
/** A merge agent that performs the real squash in the clean room (no AI). */
function squashMergeAgent(branch: string) {
return async (cwd: string): Promise<void> => {
configureIdentity(cwd);
try {
execSync(`git merge --squash ${branch}`, { cwd, stdio: "pipe" });
} catch {
// squash reported conflicts — leave them for the test's expectation.
}
// If there are unresolved conflicts, throw so landOneRepo surfaces a failure.
const unmerged = execSync("git ls-files -u", { cwd, encoding: "utf-8" }).trim();
if (unmerged.length > 0) {
throw new Error("merge conflict: unresolved paths in clean room");
}
// Nothing staged (already up to date) → leave HEAD unchanged (empty merge).
const staged = execSync("git diff --cached --name-only", { cwd, encoding: "utf-8" }).trim();
if (staged.length === 0) return;
execSync(`git commit -m "${branch}: squashed"`, { cwd, stdio: "pipe" });
};
}
const approveReviewAgent = async (): Promise<string> => "REVIEW_VERDICT: approve";
function makeTask(workspaceWorktrees: Task["workspaceWorktrees"]): Task {
return {
id: TASK_ID,
title: "Workspace merge task",
description: "",
column: "in-review",
branch: BRANCH,
dependencies: [],
steps: [],
currentStep: 0,
log: [],
workspaceWorktrees,
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
} as Task;
}
describeIfGit("landWorkspaceTask — per-repo merge loop (Phase C U1)", () => {
let fx: WorkspaceFixture;
afterEach(() => fx?.cleanup());
it("happy: both clean repos advance their OWN local integration ref with NO push", async () => {
fx = await createWorkspaceFixture(["repo-a", "repo-b"]);
addRepoBranchWithEdit(fx, "repo-a", "a feature\n");
addRepoBranchWithEdit(fx, "repo-b", "b feature\n");
const tipABefore = fx.git("repo-a", "git rev-parse refs/heads/main");
const tipBBefore = fx.git("repo-b", "git rev-parse refs/heads/main");
const store = createStore();
const task = makeTask({
"repo-a": { worktreePath: fx.repoPath("repo-a"), branch: BRANCH },
"repo-b": { worktreePath: fx.repoPath("repo-b"), branch: BRANCH },
});
const result = await landWorkspaceTask(store, task, fx.rootDir, {}, {
mergeAgent: squashMergeAgent(BRANCH),
reviewAgent: approveReviewAgent,
});
expect(result.allLanded).toBe(true);
expect(result.repos.map((r) => r.repo).sort()).toEqual(["repo-a", "repo-b"]);
for (const r of result.repos) expect(r.status).toBe("landed");
// Each repo's LOCAL integration ref advanced (main moved off its prior tip).
const tipAAfter = fx.git("repo-a", "git rev-parse refs/heads/main");
const tipBAfter = fx.git("repo-b", "git rev-parse refs/heads/main");
expect(tipAAfter).not.toBe(tipABefore);
expect(tipBAfter).not.toBe(tipBBefore);
// No remote ref / no push: the fixture repos have no remotes at all.
for (const repo of ["repo-a", "repo-b"]) {
const remotes = fx.git(repo, "git remote").trim();
expect(remotes).toBe("");
const remoteRefs = execSync("git for-each-ref refs/remotes", { cwd: fx.repoPath(repo), encoding: "utf-8" }).trim();
expect(remoteRefs).toBe("");
}
// U2 finalize-once: every repo landed → the task moves to done exactly once.
expect(store.moveTaskCalls).toEqual([{ id: TASK_ID, column: "done" }]);
expect(store.emitted.filter((e) => e.event === "task:merged")).toHaveLength(1);
});
it("per-repo resolution: each repo lands on its OWN origin/HEAD branch (override-stripping)", async () => {
fx = await createWorkspaceFixture(["repo-a", "repo-b"]);
// Give each repo a different default integration branch via a bare origin whose
// HEAD points at that branch. landWorkspaceTask strips integrationBranch/baseBranch
// overrides, so each repo resolves origin/HEAD independently.
for (const [repo, intBranch] of [["repo-a", "develop"], ["repo-b", "release"]] as const) {
const repoDir = fx.repoPath(repo);
fx.git(repo, `git branch ${intBranch}`);
const originDir = path.join(repoDir, "..", `${repo}-origin.git`);
execSync(`git init --bare ${originDir}`, { cwd: repoDir, stdio: "pipe" });
fx.git(repo, `git remote add origin ${originDir}`);
fx.git(repo, "git push origin --all");
execSync(`git symbolic-ref HEAD refs/heads/${intBranch}`, { cwd: originDir, stdio: "pipe" });
fx.git(repo, "git remote set-head origin -a");
// task branch off the integration branch with an edit
const wt = path.join(repoDir, ".wt");
fx.git(repo, `git worktree add -b ${BRANCH} ${wt} ${intBranch}`);
configureIdentity(wt);
writeFileSync(path.join(wt, "feature.txt"), `${repo} feature\n`, "utf-8");
execSync("git add feature.txt", { cwd: wt, stdio: "pipe" });
execSync(`git commit -m "feat(${TASK_ID}): add"`, { cwd: wt, stdio: "pipe" });
fx.git(repo, `git worktree remove --force ${wt}`);
}
const store = createStore();
const task = makeTask({
"repo-a": { worktreePath: fx.repoPath("repo-a"), branch: BRANCH },
"repo-b": { worktreePath: fx.repoPath("repo-b"), branch: BRANCH },
});
const result = await landWorkspaceTask(store, task, fx.rootDir, {}, {
mergeAgent: squashMergeAgent(BRANCH),
reviewAgent: approveReviewAgent,
});
expect(result.allLanded).toBe(true);
const byRepo = Object.fromEntries(result.repos.map((r) => [r.repo, r]));
expect(byRepo["repo-a"].integrationBranch).toBe("develop");
expect(byRepo["repo-b"].integrationBranch).toBe("release");
// Each landed onto its OWN integration branch's local ref.
expect(byRepo["repo-a"].status).toBe("landed");
expect(byRepo["repo-b"].status).toBe("landed");
expect(fx.git("repo-a", "git rev-parse refs/heads/develop")).toBe(byRepo["repo-a"].landedSha);
expect(fx.git("repo-b", "git rev-parse refs/heads/release")).toBe(byRepo["repo-b"].landedSha);
});
it("partial: repo B conflict → repo A lands, B reports failure, task NOT moved done", async () => {
fx = await createWorkspaceFixture(["repo-a", "repo-b"]);
addRepoBranchWithEdit(fx, "repo-a", "a feature\n");
makeConflictingRepo(fx, "repo-b");
const tipABefore = fx.git("repo-a", "git rev-parse refs/heads/main");
const store = createStore();
const task = makeTask({
"repo-a": { worktreePath: fx.repoPath("repo-a"), branch: BRANCH },
"repo-b": { worktreePath: fx.repoPath("repo-b"), branch: BRANCH },
});
const result = await landWorkspaceTask(store, task, fx.rootDir, {}, {
mergeAgent: squashMergeAgent(BRANCH),
reviewAgent: approveReviewAgent,
});
expect(result.allLanded).toBe(false);
const byRepo = Object.fromEntries(result.repos.map((r) => [r.repo, r]));
expect(byRepo["repo-a"].status).toBe("landed");
expect(byRepo["repo-b"].status).toBe("failed");
expect(byRepo["repo-b"].error).toMatch(/conflict/i);
// Repo A landed locally (its ref advanced).
expect(fx.git("repo-a", "git rev-parse refs/heads/main")).not.toBe(tipABefore);
// The task was NOT finalized/moved done on a partial land.
expect(store.moveTaskCalls).toHaveLength(0);
expect(store.emitted.some((e) => e.event === "task:merged")).toBe(false);
});
});
describe("workspace merge defense-in-depth (non-routed doors keep throwing)", () => {
it("assertNotWorkspaceTaskMerge throws WorkspaceTaskMergeError for a workspace task (store.mergeTask/aiMergeTask door)", () => {
const task = {
id: TASK_ID,
workspaceWorktrees: { "repo-a": { worktreePath: "/x/repo-a", branch: BRANCH } },
} as unknown as Task;
expect(() => assertNotWorkspaceTaskMerge(task)).toThrowError(/cannot merge until per-repo merge/i);
try {
assertNotWorkspaceTaskMerge(task);
} catch (err) {
expect((err as Error).name).toBe("WorkspaceTaskMergeError");
}
});
it("assertNotWorkspaceTaskMerge is a no-op for a single-repo task", () => {
const task = { id: TASK_ID } as unknown as Task;
expect(() => assertNotWorkspaceTaskMerge(task)).not.toThrow();
});
/*
FNXC:Workspace 2026-06-22-09:30 (Phase C review B11 — exercise the REAL merge door, not only the helper):
Calling `assertNotWorkspaceTaskMerge` directly proves the helper, but a regression where `runAiMerge`
(the sole engine merge door, R7 chokepoint) stopped invoking it would slip through. Drive the actual
door with a minimal store whose `getTask` returns the workspace task: `runAiMerge` reads the task and
calls the guard BEFORE any git work, so it rejects with WorkspaceTaskMergeError without a real repo.
*/
it("runAiMerge (engine merge door) rejects a workspace task with WorkspaceTaskMergeError", async () => {
const workspaceTask = {
id: TASK_ID,
workspaceWorktrees: { "repo-a": { worktreePath: "/x/repo-a", branch: BRANCH } },
} as unknown as Task;
const store = {
getTask: vi.fn(async () => workspaceTask),
} as unknown as TaskStore;
await expect(runAiMerge(store, "/x", TASK_ID)).rejects.toMatchObject({
name: "WorkspaceTaskMergeError",
});
});
});

View File

@@ -6,8 +6,21 @@ sub-repo absolute path (NOT the worktree path) so two concurrent workspace tasks
contending for the SAME sub-repo are serialized. Keeping it distinct from
"executor"/"step-session" means it does not collide with the executor's later
session registration on the produced worktree path.
FNXC:Workspace 2026-06-22-02:10 (Phase C U3, KTD4):
"workspace-repo-land" is a DISTINCT registry kind for the LAND-time (merge phase)
same-sub-repo lease. Like the acquire kind it is keyed by the sub-repo ABSOLUTE
path, but it guards a different lifecycle scope: two workspace tasks landing the
SAME sub-repo onto its local integration ref are serialized so their clean-room
ai-merge worktrees do not collide. This lease is for SERIALIZATION / clean-room-
collision avoidance only — it is NOT what makes the interleaved `update-ref`
correct. `advanceIntegrationBranchRef`'s CAS already makes a concurrent advance
safe by construction (concurrent-advance → rebuild). The acquire lease (execution
phase) and the land lease (merge phase) never overlap in time on the same path, so
keeping them distinct kinds (each released in its own `finally`) means a stale
entry of one kind can never be mistaken for a live hold of the other.
*/
export type ActiveSessionKind = "executor" | "step-session" | "workflow-step" | "step-session-parallel" | "ai-merge" | "workspace-repo-acquire";
export type ActiveSessionKind = "executor" | "step-session" | "workflow-step" | "step-session-parallel" | "ai-merge" | "workspace-repo-acquire" | "workspace-repo-land";
export interface ActiveSessionRegistration {
taskId: string;
@@ -43,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,

View File

@@ -203,6 +203,23 @@ export {
// FNXC:MergerUnification 2026-06-21-19:05: runAiMerge is the sole merge path
// (master-plan U0); exported for the CLI callers (fn task merge + UI-only merge).
export { runAiMerge } from "./merger-ai.js";
// FNXC:Workspace 2026-06-21-23:40 (Phase C U1): per-repo workspace merge loop +
// the extracted per-repo land primitive, exported for the CLI/dashboard merge doors.
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,
type LandRepoContext,
} from "./merger-ai.js";
export {
resolveMergePolicy,
type ResolvedMergePolicy,

File diff suppressed because it is too large Load Diff

View File

@@ -13,7 +13,7 @@ import type {
ResearchSynthesisRequest,
ResearchSynthesisResult,
} from "@fusion/core";
import { allowsAutoMergeProcessing, assertNotWorkspaceTaskMerge, 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 } 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";
@@ -121,13 +121,27 @@ function formatErrorDetails(error: unknown): { message: string; detail: string }
return { message: detail, detail };
}
/*
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,
shouldRetry: autoResolveOk && currentRetries + 1 < maxAutoMergeRetries,
maxAutoMergeRetries,
nextRetryCount: currentRetries + 1,
};
@@ -344,6 +358,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
@@ -1840,6 +1867,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
@@ -1864,6 +1904,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,
@@ -2006,6 +2047,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
@@ -2302,17 +2344,64 @@ export class ProjectEngine {
this.activeMergeSession = session;
},
};
// FNXC:Workspace 2026-06-21-19:40:
// R7 merge-boundary guard (master-plan U0). Reject workspace-mode
// tasks BEFORE any git work — they need the per-repo merge loop that
// lands in master-plan U6 (which removes this guard). Load the task
// here so the dispatch shares the one predicate in @fusion/core.
// This door is a FAST-FAIL only: a getTask failure is swallowed to null
// and the guard is skipped, but the unconditional chokepoint guard inside
// runAiMerge (which re-reads the task) is the authoritative enforcement,
// so a transient read failure here cannot let a workspace task reach git work.
// FNXC:Workspace 2026-06-21-23:40 (Phase C U1, KTD2):
// Engine merge dispatch door. A workspace-mode task (non-empty
// `workspaceWorktrees`) routes to the per-repo merge loop
// `landWorkspaceTask` (Phase C U1) instead of the singular runAiMerge —
// each sub-repo lands on its own LOCAL integration ref, no push. The
// U0 R7 throw is REPLACED by this routing (the runAiMerge chokepoint
// + store.mergeTask/aiMergeTask keep throwing as defense-in-depth).
// FAST-FAIL note preserved: a getTask failure is swallowed to null and
// 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);
if (mergeTask) assertNotWorkspaceTaskMerge(mergeTask);
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;
// `landWorkspaceTask` records each landed `landedSha`, skips
// already-landed repos on a retry (idempotent), and on full success
// finalizes the task to `done` EXACTLY ONCE. On a PARTIAL land it does
// NOT finalize — it returns `allLanded:false`, which we surface as a
// WorkspacePartialLandError so the catch-block auto-retry consumes a
// mergeRetry and re-runs (skipping landed repos) up to MAX, then parks.
const settings = await store.getSettings().catch(() => ({}) as Settings);
const workspaceResult = await landWorkspaceTask(
store,
mergeTask!,
cwd,
{ ...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("; ");
throw new WorkspacePartialLandError(
landedCount,
failed.map((r) => r.repo),
`Workspace partial land for ${taskId}: ${landedCount} repo(s) landed, ${failed.length} failed — ${detail}`,
);
}
// Finalized to done by landWorkspaceTask; report the merge as merged so
// the success path (retry reset + branch-group promotion) runs normally.
const latest = await store.getTask(taskId).catch(() => mergeTask!);
const anyLanded = workspaceResult.repos.some((r) => r.status === "landed");
return {
task: latest ?? mergeTask!,
branch: mergeTask!.branch ?? "",
merged: anyLanded,
noOp: !anyLanded,
ok: true,
commitSha: workspaceResult.repos.find((r) => r.status === "landed")?.landedSha,
mergeConfirmed: anyLanded,
worktreeRemoved: false,
branchDeleted: false,
} as MergeResult;
}
// FNXC:MergerUnification 2026-06-21-19:05:
// Master-plan U0 collapsed the merge dispatch: `runAiMerge` (the
@@ -2358,6 +2447,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);
}
@@ -2409,6 +2501,168 @@ 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).
*/
/*
FNXC:Workspace 2026-06-22-09:30 (Phase C review B7b — manual-merge busy must NOT burn mergeRetries):
A manual merge (hasManualResolver) that hits sub-repo land contention is the SAME transient
lease contention as the auto path, NOT a real land failure. Without this branch it falls
through to the generic handler below, which increments the persisted `mergeRetries` quota —
so a user mashing the merge button during contention could exhaust retries before any real
land attempt. Reject the resolver so the busy error surfaces to the user (they can retry),
WITHOUT consuming a mergeRetry. No re-enqueue: manual merges are user-driven, not engine-timed.
*/
if (err instanceof WorkspaceRepoLandBusyError && hasManualResolver) {
await store
.logEntry(taskId, `Workspace sub-repo land busy (contention): ${errorMsg}`, "WorkspaceRepoLandBusy")
.catch(() => undefined);
this.rejectMergeResolvers(taskId, err instanceof Error ? err : new Error(errorMsg));
continue;
}
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 capped exponential backoff up to the
// existing MAX (resolveMaxAutoMergeRetries), then OPERATOR-PARKS (status:"failed")
// — 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) {
/*
FNXC:Workspace 2026-06-22-09:30 (Phase C review B8 — clear stale busy quota on real outcome):
Reaching a REAL partial land means the prior transient busy contention is over. The
`workspaceBusyReenqueues` counter is otherwise only cleared on success or busy-cap
exhaustion, so a few transient busy failures followed by a real partial land would leave
a stale count — later UNRELATED contention would then resume from it and park the task
early. Clear it here so each fresh contention episode gets the full busy budget.
*/
this.workspaceBusyReenqueues.delete(taskId);
const wsSettings = await store.getSettings().catch(() => null);
const wsTask = await store.getTask(taskId).catch(() => null);
/*
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) {
/*
FNXC:Workspace 2026-06-22-09:30 (Phase C review B9 — persist retry count BEFORE arming the timer):
The retry-count write must succeed before we schedule the retry. A swallowed
`.catch(() => undefined)` here armed the timer even when the `mergeRetries` increment
never landed — so the next attempt re-read the OLD `mergeRetries` and could loop without
consuming budget, defeating the fail-closed DB-outage guard above. FAIL CLOSED: if the
write throws, park as failed (best-effort) and do NOT schedule a retry storm against a
non-responsive DB; the cooldown sweep re-evaluates once the DB recovers.
*/
try {
await store.updateTask(taskId, { mergeRetries: decision.nextRetryCount, status: null });
} catch (persistErr: unknown) {
const pmsg = persistErr instanceof Error ? persistErr.message : String(persistErr);
runtimeLog.error(
`Auto-merge: ${taskId} workspace partial land retry NOT scheduled — mergeRetries could not be persisted (DB outage?), failing closed instead of a retry storm: ${pmsg}`,
);
await store
.updateTask(taskId, { status: "failed", error: errorMsg })
.catch(() => undefined);
continue;
}
// 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)`,
);
setTimeout(() => {
if (!this.shuttingDown) this.internalEnqueueMerge(taskId);
}, delayMs);
} else {
await store
.updateTask(taskId, { status: "failed", mergeRetries: decision.maxAutoMergeRetries, error: errorMsg })
.catch(() => undefined);
await store
.logEntry(
taskId,
`Workspace partial land exhausted ${decision.maxAutoMergeRetries} retries — parking as failed for operator intervention (landed repos remain landed locally): ${errorMsg}`,
"WorkspacePartialLand",
)
.catch(() => undefined);
runtimeLog.error(
`Auto-merge: ${taskId} workspace partial land exhausted ${decision.maxAutoMergeRetries} retries — parked as failed`,
);
}
continue;
}
runtimeLog.error(`${hasManualResolver ? "Manual" : "Auto"}-merge failed for ${taskId}: ${errorMsg}`);
// Surface every merge failure on the task log so the dashboard shows