From 9fa8b386ee17c69be2517df526f1c666faae50f5 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Sat, 15 Aug 2026 03:22:13 -0700 Subject: [PATCH] FN-9059: add durable workspace coordination leases Prevent overlapping multi-node workspace operations and duplicate repository landings. - Add durable coordination leases, fence tokens, and land-intent persistence. - Fence workspace merge dispatches and repository publication across engine nodes. - Reconcile expired coordination state safely and cover lease lifecycle behavior. Files changed: .../fn-9059-workspace-durable-coordination.md | 7 + AGENTS.md | 1 + docs/architecture.md | 2 + docs/multi-project.md | 48 ++++ .../workspace-coordination-leases.pg.test.ts | 56 ++++ .../__tests__/postgres/workspace-leases.pg.test.ts | 112 ++++++++ packages/core/src/engine-node-identity.ts | 22 ++ packages/core/src/index.ts | 3 + .../0060_fn_9059_workspace_coordination_leases.sql | 10 + packages/core/src/postgres/schema-applier.ts | 13 +- packages/core/src/postgres/schema/project.ts | 31 ++ packages/core/src/store.ts | 18 ++ packages/core/src/task-store/workspace-leases.ts | 261 +++++++++++++++++ packages/core/src/tasks/workspace-lease-types.ts | 24 ++ .../engine/src/__tests__/project-engine.test.ts | 69 ++++- .../src/__tests__/self-healing-workspace.test.ts | 63 ++++- .../workspace-durable-coordination.test.ts | 49 ++++ .../src/__tests__/workspace-merger-lease.test.ts | 312 ++++++++++++++++++++- packages/engine/src/merge/merger-ai.ts | 277 ++++++++++++++++-- packages/engine/src/merge/workspace-fence-ref.ts | 171 +++++++++++ packages/engine/src/project-engine.ts | 175 ++++++++++-- packages/engine/src/runtimes/in-process-runtime.ts | 4 +- packages/engine/src/self-healing.ts | 191 ++++++++++++- packages/engine/src/util/run-audit.ts | 2 + .../engine/src/worktree/worktree-acquisition.ts | 43 ++- 25 files changed, 1901 insertions(+), 63 deletions(-) Fusion-Task-Id: FN-9059 Fusion-Task-Lineage: e51e3f54-69ee-4337-a218-4895d87474aa Co-authored-by: Fusion (runfusion.ai) --- .../fn-9059-workspace-durable-coordination.md | 7 + AGENTS.md | 1 + docs/architecture.md | 2 + docs/multi-project.md | 48 +++ .../workspace-coordination-leases.pg.test.ts | 56 ++++ .../postgres/workspace-leases.pg.test.ts | 112 +++++++ packages/core/src/engine-node-identity.ts | 22 ++ packages/core/src/index.ts | 3 + ..._fn_9059_workspace_coordination_leases.sql | 10 + packages/core/src/postgres/schema-applier.ts | 13 +- packages/core/src/postgres/schema/project.ts | 31 ++ packages/core/src/store.ts | 18 + .../core/src/task-store/workspace-leases.ts | 261 +++++++++++++++ .../core/src/tasks/workspace-lease-types.ts | 24 ++ .../src/__tests__/project-engine.test.ts | 69 +++- .../__tests__/self-healing-workspace.test.ts | 63 +++- .../workspace-durable-coordination.test.ts | 49 +++ .../__tests__/workspace-merger-lease.test.ts | 312 +++++++++++++++++- packages/engine/src/merge/merger-ai.ts | 277 ++++++++++++++-- .../engine/src/merge/workspace-fence-ref.ts | 171 ++++++++++ packages/engine/src/project-engine.ts | 175 ++++++++-- .../engine/src/runtimes/in-process-runtime.ts | 4 +- packages/engine/src/self-healing.ts | 191 ++++++++++- packages/engine/src/util/run-audit.ts | 2 + .../src/worktree/worktree-acquisition.ts | 43 ++- 25 files changed, 1901 insertions(+), 63 deletions(-) create mode 100644 .changeset/fn-9059-workspace-durable-coordination.md create mode 100644 packages/core/src/__tests__/postgres/workspace-coordination-leases.pg.test.ts create mode 100644 packages/core/src/__tests__/postgres/workspace-leases.pg.test.ts create mode 100644 packages/core/src/engine-node-identity.ts create mode 100644 packages/core/src/postgres/migrations/0060_fn_9059_workspace_coordination_leases.sql create mode 100644 packages/core/src/task-store/workspace-leases.ts create mode 100644 packages/core/src/tasks/workspace-lease-types.ts create mode 100644 packages/engine/src/__tests__/workspace-durable-coordination.test.ts create mode 100644 packages/engine/src/merge/workspace-fence-ref.ts diff --git a/.changeset/fn-9059-workspace-durable-coordination.md b/.changeset/fn-9059-workspace-durable-coordination.md new file mode 100644 index 0000000000..1cb7cc0693 --- /dev/null +++ b/.changeset/fn-9059-workspace-durable-coordination.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Prevent multi-node workspace operations from overlapping or double-landing shared repositories. +category: fix +dev: Adds migration 0060 lease and land-intent tables, FUSION_NODE_ID plus process incarnation ownership, resource fence tokens and one-publish-per-tenancy refs under refs/fusion/workspace-lease/* and refs/fusion/merge-dispatch/*. Merge-dispatch tenancy pins publish on every target sub-repository remote before any workspace land begins; merge and land commit points use fence-validated target/fence CAS operations. `isMergePending` consults durable dispatch leases after local state, while startup and periodic sweeps conservatively retire only expired leases. Pending land intents recover project-wide from remote reachability through holder or no-live-lease recovery authority. diff --git a/AGENTS.md b/AGENTS.md index 4b0a1ef1a2..e8e7b8dab3 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -306,6 +306,7 @@ Scoped exception (FN-5819/FN-8823): while project auto-merge is On, shared-branc - Workspace (Phase D U1): self-healing emits `task:reconcile-workspace-partial-land` when it re-enqueues a partial/zero-landed workspace task's per-repo land (or parks it `failed` for proven branch absence or exhausted `evidence-unavailable` branch reads), and `task:reconcile-workspace-partial-land-no-action` when `autoMerge:false`, user-pause, a live sub-repo worktree (workspace-aware liveness), or `evidence-unavailable` blocks that backward move. The bounded evidence-exhaustion reason is `evidence-unavailable-exhausted`; audit metadata remains ids/counts/outcomes-only. - Workspace (Phase D U1): self-healing emits `task:reclaim-phantom-workspace-land-lease` when it clears a leaked `workspace-repo-land` lease whose owning task is terminal/dead and older than the FN-6736 staleness floor. Archived-role and soft-deleted owners are terminal; live merging, executing, or merge-pending owners are untouched. - FN-9058: `worktree:workspace-main-checkout-edit` records workspace completion guard evidence with ids/counts/fixed outcomes only: task/repo IDs, file/commit counts, evidence or warning reason enum, `taskDoneRetryCount`, and `blocked`/`warned`/`skipped`; never paths, file content, or commit prose. +- FN-9059: workspace coordination emits `workspace-lease:*` events for lease acquisition, renewal, release, `fence-published`, `fence-superseded`, `reclaimed`, and `reclaim-refused`, plus `workspace-land-intent:*` events for write-ahead intent lifecycle and `resolve-refused`. Metadata is ids, SHAs, counts, and fixed outcomes only; it never includes a credential-bearing remote URL. - FN-9056: self-healing emits `task:reconcile-orphaned-workspace-worktree` when it reclaims a complete-lane or conservatively-idle failed/soft-deleted workspace entry. It vetoes raw/canonical active paths, task-session/executor/merge liveness, pauses and scheduled recovery; archived rows remain archive-lifecycle-owned. It runs `git worktree prune` even for already-gone paths and deletes only safely-discardable canonical `fusion/` branches. Duplicate, foreign, unowned, or outside-root claims are skipped without git work; one entry-scoped `MAX_STARVATION_DROPS` budget plus settlement bounds retries. Metadata is ids/counts/fixed outcomes: task/repo/path, success/reason/lane, worktree/prune/branch outcomes, and attempt. - FN-8144: archive emits `archive-workspace-worktree-disposer-missing` when a workspace archive has no store-scoped backend disposer; per-repository archive removal is awaited under canonical-path reservations, with failed paths quarantined for successor reconciliation. - FN-7514: the planner overseer's per-task oversight loop (`PlannerRecoveryController.tick`) emits `overseer:oversight-withheld-human-control` when the pure `evaluateOverseerHumanControl` guard withholds ALL oversight action (no steering, retry, targeted-fix, or pending confirmation) for a task that is user-paused (`task.userPaused===true`, or `task.paused===true` with no `pausedReason`) or ineligible for auto-merge processing per `allowsAutoMergeProcessing` (`autoMerge:false`/PR-based human-review terminal contract). The guard runs BEFORE FN-7513's confirmation classification, so a withheld task never records a pending confirmation. Metadata: `{ taskId, reason: "user-paused" | "auto-merge-off-human-review", stage, oversightLevel }`; deduped per (taskId, withheld reason) so it is not re-emitted every poll while the reason is unchanged. diff --git a/docs/architecture.md b/docs/architecture.md index b84c7e7279..3a76c80126 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -628,6 +628,8 @@ See [Memory Plugin Contract](./memory-plugin-contract.md) for the full plan. - **Executor**: `TaskExecutor` (`executor.ts`) implements tasks in worktrees - **Workspace acquisition shape:** per-repo acquisition persists only `workspaceWorktrees`; it never exposes a sub-repo path or branch through singular `task.worktree`/`task.branch`, including if the final workspace-state write fails. This preserves workspace classification for dashboard rendering, self-healing, and executor dispatch. - **Workspace entry mutation (FN-9052):** every per-repository `workspaceWorktrees` update goes through `TaskStore.mergeWorkspaceWorktreeEntry`, which holds the per-task PostgreSQL advisory transaction lock and merges one key under the composite project/task scope. Per-repo callers must never wholesale-replace the map, because a concurrent sibling acquisition, landing, failure, or teardown mutation would otherwise lose its entry. + - **Durable multi-node workspace coordination (FN-9059):** `project.workspace_coordination_leases`, serialized under a project/resource advisory transaction lock, owns sub-repository acquire exclusivity, per-repository land, workspace liveness/recovery guards, and merge-dispatch admission across engine processes. The owner triple `(taskId, nodeId, incarnationId)` makes only same-process re-entry idempotent; the same task from another node or process incarnation contends. Fresh/reclaimed claims receive a monotonically increasing fence token, while re-entry and renewal retain it. Git-writing tenancies publish once per tenancy—on acquire/reclaimed-expired, not re-entry—the per-repository `refs/fusion/workspace-lease/` or per-merge `refs/fusion/merge-dispatch/` fence ref. A workspace merge dispatches its deterministic per-merge pin to every target sub-repository remote before any workspace land begins; the workspace root need not be a git checkout and is never used as a substitute remote. A re-entrant claimant reuses its pin, except to repair a claim-to-publish gap with no pin; acquire-kind leases have no fence ref. A land push atomically CASes the target ref observed by the tenancy plus its repository fence and enclosing merge-dispatch fence refs in one push; a merge-only push CASes its target and dispatch pin. Thus a superseded owner is rejected even when the target tip has not moved. Lease-protected durable writes run through `withValidWorkspaceLeaseAsync`, and lease mutation is owner-and-fence conditional; no validate-then-act path is valid. + - **Workspace merge and land crash boundaries (FN-9059):** merge dispatch claims at body dispatch rather than enqueue; a losing claimant benignly drops. The body re-proves its fence at admission, the target-plus-fence atomic push, subsequent PR/branch/status effects, and terminal outcome persistence. Non-CASable effects are idempotent and follow the fenced push, so mid-merge expiry stops a superseded body at its next commit point; a pushed result whose outcome cannot persist is `merge-completed-unrecorded`, never a re-push. Before workspace land pushes, `project.workspace_land_intents` records the expected tip, intended SHA, remote/integration identity, and fence pin. The ordered protocol is intent → atomic push → lease-validated `landedSha` persist plus intent resolution. The node-independent reconciler lists pending intents project-wide, fetches the recorded remote, and resolves reachability on its integration ref. Only a live holder (its own/equal fence or a lower-fence predecessor) or recovery with no held unexpired lease can resolve an intent; stale fence matches, local tip equality, and local object availability are not authority. - **Main-checkout completion guard (FN-9058):** `fn_task_done` probes every configured sub-repo main checkout before any workspace worktree invariant, so `main_checkout_edit` takes precedence over `no_commits` and cannot be skipped by zero-acquire or no-commit eligibility. It uses the immutable first-execution anchor (never only the re-stamped per-attempt timestamp), blocks task-era status entries and bounded recent-HEAD evidence without `--since` or ancestry filtering, and emits `worktree:workspace-main-checkout-edit`. Unattributable pre-existing dirt, unavailable probes, and unresolved timing only warn: refusal has a bounded requeue budget and the guard is read-only. - **Task-pinned orphan recovery:** task-ID-pinned acquisition holds one path reservation across classification, preservation, quarantine reconciliation, and recreation. Inactive incomplete or unregistered directories are atomically moved to `/.fusion/recovery/worktrees`, or to `/.fusion-recovery/worktrees` after an `EXDEV` cross-filesystem refusal. Each actual recovery root retains the newest 10 recognized Fusion-generated entries; pruning is fail-soft and preserves unknown, symlinked, unreadable, or active paths. Worktree pool and self-healing scans exclude both `.ai-merge` and `.fusion-recovery` as internal container boundaries. diff --git a/docs/multi-project.md b/docs/multi-project.md index 89ef758fb7..c292d8709f 100644 --- a/docs/multi-project.md +++ b/docs/multi-project.md @@ -82,6 +82,54 @@ What is **not** multi-node via shared DB alone: Canonical ownership / control-plane contract: [`docs/shared-mesh-protocol.md`](./shared-mesh-protocol.md). +## Durable workspace coordination across nodes + +Workspace mode uses project-scoped PostgreSQL coordination records, not a node's in-memory registry, for sub-repository acquisition, per-repository landing, workspace-task liveness, merge-pending recovery guards, and merge-body dispatch. The durable records are `project.workspace_coordination_leases` and the write-ahead `project.workspace_land_intents`; lease operations are serialized with a project/resource advisory transaction lock. + +Set a stable `FUSION_NODE_ID` on every engine host. Each process also creates a fresh, process-local incarnation ID at startup. A lease owner is the triple `(ownerTaskId, ownerNodeId, ownerIncarnationId)`, with a monotonically increasing fence token: + +| Existing owner vs claimant | Result | +| --- | --- | +| Same task, node, and incarnation | Re-entrant claim: retain the fence token and refresh the TTL. | +| Same task, different node or incarnation | Contention: fail closed; the claimant must wait for expiry or authorized reclamation. | +| Different task | Contention: fail closed. | +| Expired holder | A new claimant may atomically reclaim the record and receives a new fence token. | + +The TTL and its renewal timer indicate liveness; they are not authorization to mutate a shared resource. A caller must prove its owner triple and fence token at the action boundary. Lease-row changes use owner-and-fence-scoped conditional updates, while durable land intent, `landedSha`, merge admission, and merge-outcome writes run inside the same advisory-locked transaction that re-verifies the lease. Do not validate a lease and then act outside that transaction. + +### Resource-bound fences + +The fence token is enforced by the resource as well as by the database. Fusion uses exactly these mechanisms: + +1. **Git fence refs.** A sub-repository land publishes `refs/fusion/workspace-lease/` and a merge body publishes `refs/fusion/merge-dispatch/`. At the irreversible push, one `git push --atomic` compare-and-swaps the target ref observed by that tenancy and every applicable published pin: the repository fence plus the enclosing merge-dispatch fence for workspace land, or the dispatch pin for a merge-only push. A target-tip-only CAS is insufficient: a superseded owner can otherwise push while the target tip is still unchanged. The remote must permit both `refs/fusion/...` namespaces; a rejected namespace or fence-ref publication fails closed and never falls back to a tip-only push. +2. **Transaction-bound durable writes.** The lease validity check and the protected state write occur in one advisory-locked transaction. +3. **Owner-and-fence conditional lease updates.** Renew, release, and lease-state transitions cannot alter a successor's row. + +A fence ref is published once when a git-writing tenancy is `acquired` or `reclaimed-expired`. A re-entrant claim reuses the existing `fenceRefName` and `fenceRefSha`; it does not rotate the pin or increment the fence. Republishing would invalidate that same tenancy's prepared CAS push and orphan its pending intent. The only re-entrant exception is the publish gap: a git-writing row with no fence ref because the process died after claiming but before publishing may publish its missing pin. For a workspace merge-dispatch tenancy, the same deterministic pin is published to every target sub-repository remote before any workspace land begins; the non-git workspace root is never treated as the protected remote. Acquire-kind leases never carry a fence ref, and renewal never publishes or changes one. + +### Merge dispatch and commit points + +A merge-dispatch lease is claimed when the queue dispatches a merge body, not when it enqueues one. A losing dispatch claim is a benign drop rather than a task failure; the enqueue-to-dispatch interval intentionally has no active merge body. + +The merge body re-proves its fence at every commit point: dispatch admission; the atomic target-plus-dispatch-fence push (and repository fence where a workspace land also owns one); each subsequent PR merge, remote-branch deletion, or status effect; and terminal outcome persistence. The fenced push is the first shared irreversible effect. Non-CASable remote effects must be idempotent and occur after it. Therefore a lease may expire during a long merge without making a second push or outcome valid: a superseded body stops before its next commit point. If the push succeeded but a later outcome write is fenced out, Fusion reports `merge-completed-unrecorded`; operators should inspect the remote target and audit trail rather than retrying blindly or expecting a rollback. + +### Crash-safe workspace land recovery + +Before a land push, Fusion writes a durable pending land intent containing the task/repository identity, expected target tip, intended SHA, `remoteUrl`, integration ref, and fence-ref pin. The order is intent → atomic target-and-applicable-fences push → one lease-validated transaction that persists `landedSha` and resolves the intent. This closes the crash window between a successful push and durable task state. + +Pending intents are enumerated project-wide from PostgreSQL, not from the surviving node's local candidates, registries, or worktrees. Recovery fetches the recorded remote and proves intended-SHA reachability on the recorded integration ref; local tip equality, TTL alone, or a stale local object store are not proof. A subsequent land first resolves any pending intent for its task/repository, so it cannot squash a second time across an unresolved prior commit point. + +Exactly two authorities may resolve an intent: + +- The holder-authorized resolver holds a live lease handle and may resolve its equal-fence intent or a strictly lower-fence predecessor. +- The recovery-authorized resolver has no handle and resolves only when its transaction proves no held, unexpired lease exists; it also refuses an intent/lease fence mismatch. + +There is no third writer. A pending intent is an operator-visible recovery state: inspect its audit records and remote integration ref, then allow holder or recovery reconciliation to establish ground truth rather than manually assuming the land failed. + +### Reclamation and operator behavior + +Lease renewal extends a live holder, but restart recovery never clears an unexpired lease—not even for a duplicated `FUSION_NODE_ID`. This deliberately trades a bounded TTL wait for safety. Startup releases only this node's expired predecessor-incarnation rows; periodic maintenance marks expired rows reclaimable without changing intents or fence refs. `isMergePending` first checks local queue state, then queries held `merge-dispatch` leases for every node; a durable-store error is conservatively pending. The phantom land-lease sweep also enumerates durable `land` and `acquire` rows, then reclaims only through an owner-and-fence CAS that derives terminal ownership from the task row in the same transaction; callers cannot supply a stale terminal proof. On contention, Fusion reports a busy/deferred operation and preserves the incumbent holder. Operators should wait for normal release/expiry or investigate a persistently pending intent, instead of bypassing a fence or editing a shared checkout. + ### Cluster membership and process ownership - Topology visibility is cluster-wide: dashboard mesh reads aggregate node registry state (and optional remote health probes), with degraded fallback metadata when a peer HTTP probe fails. diff --git a/packages/core/src/__tests__/postgres/workspace-coordination-leases.pg.test.ts b/packages/core/src/__tests__/postgres/workspace-coordination-leases.pg.test.ts new file mode 100644 index 0000000000..7babd65f30 --- /dev/null +++ b/packages/core/src/__tests__/postgres/workspace-coordination-leases.pg.test.ts @@ -0,0 +1,56 @@ +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { createSharedPgTaskStoreTestHarness, pgDescribe, type SharedPgTaskStoreHarness } from "../../__test-utils__/pg-test-harness.js"; + +const pgTest = pgDescribe; + +/* +FNXC:WorkspaceMergeDispatch 2026-08-15-09:46: +A merge body that already pushed may wake after a successor has reclaimed its dispatch tenancy. +The terminal callback must be guarded by the durable owner-and-fence transaction, not by a renewal +callback or an in-process belief, so a stale body cannot mark the task complete. +*/ +pgTest("workspace merge dispatch finalization (PostgreSQL)", () => { + const h: SharedPgTaskStoreHarness = createSharedPgTaskStoreTestHarness({ prefix: "fusion_workspace_dispatch_finalization" }); + + beforeAll(h.beforeAll); + beforeEach(h.beforeEach); + afterEach(h.afterEach); + afterAll(h.afterAll); + + it("does not invoke a predecessor terminal callback after a durable successor reclaim", async () => { + const predecessorStore = h.store(); + const successorStore = h.store(); + const task = await predecessorStore.createTask({ description: "durable merge terminal fence" }); + const predecessor = await predecessorStore.acquireWorkspaceLease({ + leaseKey: "merge-dispatch:FN-9059-finalize", + kind: "merge-dispatch", + owner: { taskId: "FN-9059-finalize", nodeId: "node-b", incarnationId: "inc-b" }, + leaseMs: 1, + }); + if (predecessor.outcome === "conflict") throw new Error("expected predecessor lease"); + + // The renewal callback never runs. Once the durable TTL passes, another node reclaims with a + // higher token and the predecessor's finalization transaction must refuse its old handle. + await new Promise((resolve) => setTimeout(resolve, 5)); + const successor = await successorStore.acquireWorkspaceLease({ + leaseKey: predecessor.handle.leaseKey, + kind: "merge-dispatch", + owner: { taskId: "FN-9059-finalize", nodeId: "node-a", incarnationId: "inc-a" }, + leaseMs: 60_000, + }); + expect(successor.outcome).toBe("reclaimed-expired"); + + /* + FNXC:WorkspaceMergeDispatch 2026-08-15-10:18: + Model the post-push terminal persist with the real PostgreSQL TaskStore. A successor reclaim + must prevent the callback itself from entering its durable task mutation, not merely reject a + mocked finalizer after a renewal timer notices expiry. + */ + const terminalWrite = vi.fn(async () => { + await predecessorStore.updateTask(task.id, { title: "stale merge outcome" }); + }); + await expect(predecessorStore.withValidWorkspaceLease(predecessor.handle, terminalWrite)).rejects.toThrow("Workspace lease is no longer valid"); + expect(terminalWrite).not.toHaveBeenCalled(); + expect((await predecessorStore.getTask(task.id))?.title).toBe(task.title); + }); +}); diff --git a/packages/core/src/__tests__/postgres/workspace-leases.pg.test.ts b/packages/core/src/__tests__/postgres/workspace-leases.pg.test.ts new file mode 100644 index 0000000000..a00346de6a --- /dev/null +++ b/packages/core/src/__tests__/postgres/workspace-leases.pg.test.ts @@ -0,0 +1,112 @@ +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from "vitest"; +import { createSharedPgTaskStoreTestHarness, pgDescribe, type SharedPgTaskStoreHarness } from "../../__test-utils__/pg-test-harness.js"; +import type { WorkspaceLeaseOwner } from "../../tasks/workspace-lease-types.js"; + +const pgTest = pgDescribe; + +/* +FNXC:WorkspaceLease 2026-08-15-08:57: +A workspace lease is reentrant only for its complete task/node/incarnation owner +triple. Reentry must renew its TTL without minting a fence token or dropping a +published fence pin; either change lets a live writer lose its git CAS authority. +*/ +pgTest("workspace coordination leases (PostgreSQL)", () => { + const h: SharedPgTaskStoreHarness = createSharedPgTaskStoreTestHarness({ prefix: "fusion_workspace_leases" }); + + beforeAll(h.beforeAll); + beforeEach(h.beforeEach); + afterEach(h.afterEach); + afterAll(h.afterAll); + + const owner: WorkspaceLeaseOwner = { + taskId: "FN-9059-owner", + nodeId: "node-a", + incarnationId: "incarnation-a", + }; + + it("conflicts when any owner-triple member differs, but reenters for the exact owner", async () => { + const first = h.store(); + const second = h.store(); + const leaseKey = "workspace:repo-a:owner-triple"; + const acquired = await first.acquireWorkspaceLease({ leaseKey, kind: "land", owner, leaseMs: 60_000 }); + expect(acquired.outcome).toBe("acquired"); + if (acquired.outcome === "conflict") throw new Error("expected initial workspace lease claim"); + + const changes: Array> = [ + { taskId: "FN-9059-other" }, + { nodeId: "node-b" }, + { incarnationId: "incarnation-b" }, + ]; + for (const changedOwner of changes) { + const result = await second.acquireWorkspaceLease({ + leaseKey, + kind: "land", + owner: { ...owner, ...changedOwner }, + leaseMs: 60_000, + }); + expect(result.outcome).toBe("conflict"); + if (result.outcome === "conflict") { + expect(result.conflict).toMatchObject({ + taskId: owner.taskId, + nodeId: owner.nodeId, + incarnationId: owner.incarnationId, + fenceToken: acquired.handle.fenceToken, + }); + } + } + + const reentrant = await second.acquireWorkspaceLease({ leaseKey, kind: "land", owner, leaseMs: 60_000 }); + expect(reentrant.outcome).toBe("reentrant"); + if (reentrant.outcome === "conflict") throw new Error("expected exact owner reentry"); + expect(reentrant.handle.fenceToken).toBe(acquired.handle.fenceToken); + }); + + it("labels a released row as reclaimed and clears its previous fence pin", async () => { + const store = h.store(); + const leaseKey = "workspace:repo-a:released-reclaim"; + const first = await store.acquireWorkspaceLease({ leaseKey, kind: "land", owner, leaseMs: 60_000 }); + if (first.outcome === "conflict") throw new Error("expected initial claim"); + const pinned = await store.recordWorkspaceLeaseFenceRef({ + handle: first.handle, + fenceRefName: "refs/fusion/fence/released", + fenceRefSha: "fence-sha", + }); + await store.releaseWorkspaceLease(pinned); + + const reclaimed = await store.acquireWorkspaceLease({ leaseKey, kind: "land", owner, leaseMs: 60_000 }); + expect(reclaimed.outcome).toBe("reclaimed-expired"); + if (reclaimed.outcome === "conflict") throw new Error("expected reclaimed lease"); + expect(reclaimed.handle.fenceToken).toBeGreaterThan(pinned.fenceToken); + expect(reclaimed.handle.fenceRefName).toBeUndefined(); + expect(reclaimed.handle.fenceRefSha).toBeUndefined(); + }); + + it("preserves a land fence pin across reentrant renewal", async () => { + const store = h.store(); + const leaseKey = "workspace:repo-a:fence-pin"; + const acquired = await store.acquireWorkspaceLease({ leaseKey, kind: "land", owner, leaseMs: 60_000 }); + expect(acquired.outcome).toBe("acquired"); + if (acquired.outcome === "conflict") throw new Error("expected initial workspace lease claim"); + + const pinned = await store.recordWorkspaceLeaseFenceRef({ + handle: acquired.handle, + fenceRefName: "refs/fusion/fence/FN-9059", + fenceRefSha: "fence-sha", + }); + const reentrant = await store.acquireWorkspaceLease({ leaseKey, kind: "land", owner, leaseMs: 60_000 }); + + expect(reentrant).toMatchObject({ + outcome: "reentrant", + handle: { + fenceToken: pinned.fenceToken, + fenceRefName: pinned.fenceRefName, + fenceRefSha: pinned.fenceRefSha, + }, + }); + const [persisted] = await store.inspectWorkspaceLeases({ leaseKeys: [leaseKey] }); + expect(persisted?.fenceRefName).toBe(pinned.fenceRefName); + expect(persisted?.fenceRefSha).toBe(pinned.fenceRefSha); + }); +}); + +void describe; diff --git a/packages/core/src/engine-node-identity.ts b/packages/core/src/engine-node-identity.ts new file mode 100644 index 0000000000..0aa855eda0 --- /dev/null +++ b/packages/core/src/engine-node-identity.ts @@ -0,0 +1,22 @@ +import { hostname } from "node:os"; +import { randomUUID } from "node:crypto"; + +let defaultNodeId: string | undefined; +let incarnationId: string | undefined; + +/** + * FNXC:Workspace 2026-08-15-08:23: + * FUSION_NODE_ID names a deployment slot, not an authority to clear an + * unexpired lease. Without it, hostname-pid changes after restart, so recovery + * intentionally waits for TTL; the incarnation prevents restart inheritance. + */ +export function resolveEngineNodeId(): string { + const configured = process.env.FUSION_NODE_ID?.trim(); + if (configured) return configured; + return defaultNodeId ??= `${hostname()}-${process.pid}`; +} + +/** Per-process identity distinguishes a restarted slot from its predecessor. */ +export function resolveEngineIncarnationId(): string { + return incarnationId ??= randomUUID(); +} diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index a0237a7f3f..f3c810ba8b 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -13,6 +13,9 @@ export type { ReleaseSymbolLocksResult, ReconcileStaleSymbolLocksResult, } from "./tasks/symbol-lock-types.js"; +export { resolveEngineNodeId, resolveEngineIncarnationId } from "./engine-node-identity.js"; +export { isTerminalWorkspaceLeaseOwner } from "./tasks/workspace-lease-types.js"; +export type { WorkspaceLeaseKind, WorkspaceLeaseStatus, WorkspaceLeaseOwner, WorkspaceLeaseHandle, WorkspaceLease, WorkspaceLeaseConflict, WorkspaceLeaseClaimOutcome, WorkspaceLeaseFenceOutcome, WorkspaceLandIntent, WorkspaceLandIntentResolution, WorkspaceLandIntentResolveOutcome, WorkspaceLeaseReclaimOutcome, AcquireWorkspaceLeaseResult } from "./tasks/workspace-lease-types.js"; export { normalizeSymbolLockKey, extractSymbolLockIdentity, diff --git a/packages/core/src/postgres/migrations/0060_fn_9059_workspace_coordination_leases.sql b/packages/core/src/postgres/migrations/0060_fn_9059_workspace_coordination_leases.sql new file mode 100644 index 0000000000..b12e6f875f --- /dev/null +++ b/packages/core/src/postgres/migrations/0060_fn_9059_workspace_coordination_leases.sql @@ -0,0 +1,10 @@ +-- FNXC:Workspace 2026-08-15-08:23: durable project-scoped leases fence multi-node workspace writers; intents survive a post-push process death. +CREATE TABLE IF NOT EXISTS project.workspace_coordination_leases ( + project_id text NOT NULL DEFAULT current_setting('fusion.project_id', true), lease_key text NOT NULL, kind text NOT NULL, owner_task_id text NOT NULL, owner_node_id text NOT NULL, owner_incarnation_id text NOT NULL, owner_run_id text, fence_token bigint NOT NULL DEFAULT 0, fence_ref_name text, fence_ref_sha text, status text NOT NULL, acquired_at text NOT NULL, renewed_at text NOT NULL, expires_at text NOT NULL, created_at text NOT NULL, updated_at text NOT NULL, PRIMARY KEY(project_id, lease_key), CONSTRAINT workspace_coordination_leases_kind_check CHECK(kind IN ('acquire','land','merge-dispatch')), CONSTRAINT workspace_coordination_leases_status_check CHECK(status IN ('held','released','expired'))); +CREATE INDEX IF NOT EXISTS "idxWorkspaceCoordinationLeasesOwnerTask" ON project.workspace_coordination_leases(project_id, owner_task_id); +CREATE INDEX IF NOT EXISTS "idxWorkspaceCoordinationLeasesOwnerNode" ON project.workspace_coordination_leases(project_id, owner_node_id); +CREATE INDEX IF NOT EXISTS "idxWorkspaceCoordinationLeasesExpiry" ON project.workspace_coordination_leases(status, expires_at); +CREATE TABLE IF NOT EXISTS project.workspace_land_intents ( + project_id text NOT NULL DEFAULT current_setting('fusion.project_id', true), task_id text NOT NULL, repo_rel_path text NOT NULL, remote_url text NOT NULL, integration_ref text NOT NULL, intended_sha text NOT NULL, expected_tip text NOT NULL, fence_ref_name text NOT NULL, fence_ref_sha text NOT NULL, owner_task_id text NOT NULL, owner_node_id text NOT NULL, owner_incarnation_id text NOT NULL, fence_token bigint NOT NULL, status text NOT NULL, resolved_sha text, resolution text, created_at text NOT NULL, updated_at text NOT NULL, resolved_at text, PRIMARY KEY(project_id, task_id, repo_rel_path), CONSTRAINT workspace_land_intents_status_check CHECK(status IN ('pending','recorded','abandoned')), CONSTRAINT workspace_land_intents_resolution_check CHECK(resolution IS NULL OR resolution IN ('landed','not-landed'))); +CREATE INDEX IF NOT EXISTS "idxWorkspaceLandIntentsStatus" ON project.workspace_land_intents(project_id, status); +DO $$ DECLARE tbl text; BEGIN FOREACH tbl IN ARRAY ARRAY['workspace_coordination_leases','workspace_land_intents'] LOOP EXECUTE format('ALTER TABLE project.%I ENABLE ROW LEVEL SECURITY', tbl); EXECUTE format('ALTER TABLE project.%I FORCE ROW LEVEL SECURITY', tbl); EXECUTE format('DROP POLICY IF EXISTS fusion_project_isolation ON project.%I', tbl); EXECUTE format('CREATE POLICY fusion_project_isolation ON project.%I USING (current_setting(''fusion.project_bypass'', true) = ''on'' OR project_id = current_setting(''fusion.project_id'', true)) WITH CHECK (current_setting(''fusion.project_bypass'', true) = ''on'' OR project_id = current_setting(''fusion.project_id'', true))', tbl); EXECUTE format('DROP TRIGGER IF EXISTS fusion_assign_project_id ON project.%I', tbl); EXECUTE format('CREATE TRIGGER fusion_assign_project_id BEFORE INSERT OR UPDATE OF project_id ON project.%I FOR EACH ROW EXECUTE FUNCTION project.fusion_assign_project_id()', tbl); END LOOP; END $$; diff --git a/packages/core/src/postgres/schema-applier.ts b/packages/core/src/postgres/schema-applier.ts index cb37eaff24..5da440f8ae 100644 --- a/packages/core/src/postgres/schema-applier.ts +++ b/packages/core/src/postgres/schema-applier.ts @@ -64,7 +64,8 @@ capacity-model table drop that landed while this PR was open. /* FNXC:MultiProjectIsolation 2026-08-11-10:25: schema startup must register project-local agent ratings before bound stores scope their mutations. */ /* FNXC:MessageArchive 2026-08-12-22:14: 0058 persists non-destructive mailbox archival on upgrades. */ /* FNXC:TaskRecommendations 2026-08-13-22:23: upgrades must install the source-agent index before duplicate intake queries it. */ -export const SCHEMA_BASELINE_VERSION = "0059"; +/* FNXC:WorkspaceLease 2026-08-15-12:00: the baseline ceiling must include durable coordination tables so an upgraded database is never rejected by the current binary. */ +export const SCHEMA_BASELINE_VERSION = "0060"; /** FNXC:SymbolLock 2026-07-20-10:00: upgrades need durable task declarations before admission resolves symbols. */ export const TASK_DECLARED_SYMBOLS_VERSION = "0028"; const INITIAL_SCHEMA_VERSION = "0000"; @@ -224,6 +225,7 @@ export const PROJECT_OWNERSHIP_DEFAULT_RECONCILIATION_VERSION = "0057"; /** FNXC:MessageArchive 2026-08-12-22:14: explicit registration prevents the archived-message migration from being skipped. */ export const MESSAGE_ARCHIVE_SCHEMA_VERSION = "0058"; export const TASK_SOURCE_AGENT_INDEX_VERSION = "0059"; +export const WORKSPACE_COORDINATION_LEASES_SCHEMA_VERSION = "0060"; /** SECURITY DEFINER helper that only inserts LEGACY_ADOPTION_DRAINED_MARKER. */ export const LEGACY_ADOPTION_DRAINED_MARKER_FUNCTION = "fusion_mark_legacy_adoption_drained"; @@ -457,6 +459,7 @@ const PROJECT_OWNERSHIP_DECLARATION_DRIFT_MIGRATION_PATH = join(MIGRATIONS_DIR, const PROJECT_OWNERSHIP_DEFAULT_RECONCILIATION_MIGRATION_PATH = join(MIGRATIONS_DIR, "0057_fn_9004_project_ownership_default_reconciliation.sql"); const MESSAGE_ARCHIVE_SCHEMA_MIGRATION_PATH = join(MIGRATIONS_DIR, "0058_fn_9014_message_archive.sql"); const TASK_SOURCE_AGENT_INDEX_MIGRATION_PATH = join(MIGRATIONS_DIR, "0059_fn_9037_tasks_source_agent_index.sql"); +const WORKSPACE_COORDINATION_LEASES_MIGRATION_PATH = join(MIGRATIONS_DIR, "0060_fn_9059_workspace_coordination_leases.sql"); /** * Ensure the migration bookkeeping table exists. Lives in the public schema so @@ -586,6 +589,7 @@ export async function applySchemaBaseline( const projectOwnershipDefaultReconciliationAlreadyApplied = applied.includes(PROJECT_OWNERSHIP_DEFAULT_RECONCILIATION_VERSION); const messageArchiveSchemaAlreadyApplied = applied.includes(MESSAGE_ARCHIVE_SCHEMA_VERSION); const taskSourceAgentIndexAlreadyApplied = applied.includes(TASK_SOURCE_AGENT_INDEX_VERSION); + const workspaceCoordinationLeasesAlreadyApplied = applied.includes(WORKSPACE_COORDINATION_LEASES_SCHEMA_VERSION); assertBinaryNotOlderThanDatabase(applied); let schemaChanged = false; @@ -1292,6 +1296,13 @@ export async function applySchemaBaseline( await tx.execute(sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${TASK_SOURCE_AGENT_INDEX_VERSION}) ON CONFLICT (version) DO NOTHING`); schemaChanged = true; } + /* FNXC:Workspace 2026-08-15-08:23: register 0060 explicitly; an unregistered migration never protects upgraded multi-node deployments. */ + if (!workspaceCoordinationLeasesAlreadyApplied) { + const migrationSql = await readFile(WORKSPACE_COORDINATION_LEASES_MIGRATION_PATH, "utf8"); + await tx.execute(sql.raw(migrationSql)); + await tx.execute(sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${WORKSPACE_COORDINATION_LEASES_SCHEMA_VERSION}) ON CONFLICT (version) DO NOTHING`); + schemaChanged = true; + } return { applied: schemaChanged, pluginHooksRun: pluginHooks.length }; }); } diff --git a/packages/core/src/postgres/schema/project.ts b/packages/core/src/postgres/schema/project.ts index 63c9b560fc..8364e2e174 100644 --- a/packages/core/src/postgres/schema/project.ts +++ b/packages/core/src/postgres/schema/project.ts @@ -559,6 +559,37 @@ export const distributedTaskIdReservations = projectSchema.table("distributed_ta index("idxDistributedTaskIdReservationsExpiry").on(t.status, t.expiresAt), ]); +// ── Durable workspace coordination leases ──────────────────────────── +/* +FNXC:Workspace 2026-08-15-08:23: +Acquire and land share a repo key because either tenancy must exclude another +writer. The owner triple prevents another node or restarted process using the +same task id from looking reentrant. renewedAt is audit observability only; +expiresAt is the sole liveness clock and neither is a resource fence. Tokens +matter only when a database write or git CAS enforces them. Land and +merge-dispatch retain their published fence pin so a reclaimed tenancy rejects +a stalled push even with an unchanged tip; reentry preserves it, while reclaim +clears it for the new tenancy. Acquire leases never publish a pin. +*/ +export const workspaceCoordinationLeases = projectSchema.table("workspace_coordination_leases", { + projectId: text("project_id").notNull().default(sql`current_setting('fusion.project_id', true)`), + leaseKey: text("lease_key").notNull(), kind: text("kind").notNull(), + ownerTaskId: text("owner_task_id").notNull(), ownerNodeId: text("owner_node_id").notNull(), + ownerIncarnationId: text("owner_incarnation_id").notNull(), ownerRunId: text("owner_run_id"), + fenceToken: bigint("fence_token", { mode: "bigint" }).notNull().default(sql`0`), + fenceRefName: text("fence_ref_name"), fenceRefSha: text("fence_ref_sha"), status: text("status").notNull(), + acquiredAt: text("acquired_at").notNull(), renewedAt: text("renewed_at").notNull(), expiresAt: text("expires_at").notNull(), createdAt: text("created_at").notNull(), updatedAt: text("updated_at").notNull(), +}, (t) => [ + primaryKey({ columns: [t.projectId, t.leaseKey] }), + check("workspace_coordination_leases_kind_check", sql`${t.kind} IN ('acquire', 'land', 'merge-dispatch')`), + check("workspace_coordination_leases_status_check", sql`${t.status} IN ('held', 'released', 'expired')`), + index("idxWorkspaceCoordinationLeasesOwnerTask").on(t.projectId, t.ownerTaskId), index("idxWorkspaceCoordinationLeasesOwnerNode").on(t.projectId, t.ownerNodeId), index("idxWorkspaceCoordinationLeasesExpiry").on(t.status, t.expiresAt), +]); +/* FNXC:Workspace 2026-08-15-08:23: SIGKILL between push and persist leaves this self-contained, non-TTL evidence. Recovery reads remote reachability, never memory or a local object store; only holder and orphan authorities may resolve it. Reentry preserves its pin. */ +export const workspaceLandIntents = projectSchema.table("workspace_land_intents", { + projectId: text("project_id").notNull().default(sql`current_setting('fusion.project_id', true)`), taskId: text("task_id").notNull(), repoRelPath: text("repo_rel_path").notNull(), remoteUrl: text("remote_url").notNull(), integrationRef: text("integration_ref").notNull(), intendedSha: text("intended_sha").notNull(), expectedTip: text("expected_tip").notNull(), fenceRefName: text("fence_ref_name").notNull(), fenceRefSha: text("fence_ref_sha").notNull(), ownerTaskId: text("owner_task_id").notNull(), ownerNodeId: text("owner_node_id").notNull(), ownerIncarnationId: text("owner_incarnation_id").notNull(), fenceToken: bigint("fence_token", { mode: "bigint" }).notNull(), status: text("status").notNull(), resolvedSha: text("resolved_sha"), resolution: text("resolution"), createdAt: text("created_at").notNull(), updatedAt: text("updated_at").notNull(), resolvedAt: text("resolved_at"), +}, (t) => [primaryKey({ columns: [t.projectId, t.taskId, t.repoRelPath] }), check("workspace_land_intents_status_check", sql`${t.status} IN ('pending', 'recorded', 'abandoned')`), check("workspace_land_intents_resolution_check", sql`${t.resolution} IS NULL OR ${t.resolution} IN ('landed', 'not-landed')`), index("idxWorkspaceLandIntentsStatus").on(t.projectId, t.status)]); + // ── Durable symbol locks ───────────────────────────────────────────── /* FNXC:SymbolLock 2026-07-30-14:10: diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index 42d0401b80..cb54bb0565 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -133,6 +133,8 @@ import { reconcilePhantomCommittedReservationsAsync } from "./task-store/async/a import { resolveTaskSymbolsForTask, type TaskSymbolResolution } from "./tasks/task-symbol-resolution.js"; import { acquireSymbolLocksAsync, inspectSymbolLockConflictsAsync, reconcileStaleSymbolLocksAsync, releaseSymbolLocksAsync, renewSymbolLocksAsync } from "./task-store/symbol-locks.js"; import type { AcquireSymbolLocksResult, ReconcileStaleSymbolLocksResult, ReleaseSymbolLocksResult, RenewSymbolLocksResult, SymbolLockConflict, SymbolLockOwner } from "./tasks/symbol-lock-types.js"; +import { acquireWorkspaceLeaseAsync, inspectWorkspaceLeasesAsync, listPendingWorkspaceLandIntentsAsync, reclaimWorkspaceLeaseAsync, recordWorkspaceLandIntentAsync, recordWorkspaceLeaseFenceRefAsync, reconcileExpiredWorkspaceLeasesAsync, releaseStaleWorkspaceLeasesForNodeAsync, releaseWorkspaceLeaseAsync, renewWorkspaceLeaseAsync, resolveOrphanedWorkspaceLandIntentAsync, resolveWorkspaceLandIntentAsync, validateWorkspaceLeaseFenceAsync, withValidWorkspaceLeaseAsync } from "./task-store/workspace-leases.js"; +import type { WorkspaceLeaseHandle, WorkspaceLeaseKind, WorkspaceLeaseOwner } from "./tasks/workspace-lease-types.js"; import { queryRunAuditEvents } from "./task-store/async/async-audit.js"; import { isValidMergeRequestTransitionImpl, releaseMergeQueueLeaseImpl, collectMergeDetailsImpl, applyPrMergedTransitionImpl } from "./task-store/merge-queue-ops-2.js"; import { upsertWorkflowWorkItemImpl, replaceActiveTaskWorkflowContinuationImpl, seedStrandedPlanReviewContinuationImpl, transitionWorkflowWorkItemImpl, acquireWorkflowWorkItemLeaseImpl } from "./task-store/workflow-workitems-ops-2.js"; @@ -971,6 +973,22 @@ export class TaskStore extends EventEmitter { return reconcileStaleSymbolLocksAsync(this); } + /** FNXC:Workspace 2026-08-15-08:23: Facade methods preserve the project-bound durable coordination seam. */ + acquireWorkspaceLease(input: { leaseKey: string; kind: WorkspaceLeaseKind; owner: WorkspaceLeaseOwner; leaseMs: number }) { return acquireWorkspaceLeaseAsync(this, input); } + renewWorkspaceLease(handle: WorkspaceLeaseHandle, leaseMs: number) { return renewWorkspaceLeaseAsync(this, handle, leaseMs); } + releaseWorkspaceLease(handle: WorkspaceLeaseHandle) { return releaseWorkspaceLeaseAsync(this, handle); } + withValidWorkspaceLease(handle: WorkspaceLeaseHandle, fn: (tx: import("./postgres/data-layer.js").DbTransaction) => Promise) { return withValidWorkspaceLeaseAsync(this, handle, fn); } + validateWorkspaceLeaseFence(input: { leaseKey: string; owner: WorkspaceLeaseOwner; fenceToken: bigint }) { return validateWorkspaceLeaseFenceAsync(this, input); } + recordWorkspaceLeaseFenceRef(input: { handle: WorkspaceLeaseHandle; fenceRefName: string; fenceRefSha: string }) { return recordWorkspaceLeaseFenceRefAsync(this, input); } + inspectWorkspaceLeases(filter: { taskId?: string; leaseKeys?: string[] } = {}) { return inspectWorkspaceLeasesAsync(this, filter); } + reclaimWorkspaceLease(input: Parameters[1]) { return reclaimWorkspaceLeaseAsync(this, input); } + reconcileExpiredWorkspaceLeases() { return reconcileExpiredWorkspaceLeasesAsync(this); } + releaseStaleWorkspaceLeasesForNode(nodeId: string, options: { currentIncarnationId: string }) { return releaseStaleWorkspaceLeasesForNodeAsync(this, nodeId, options); } + recordWorkspaceLandIntent(input: Parameters[1]) { return recordWorkspaceLandIntentAsync(this, input); } + listPendingWorkspaceLandIntents(filter: Parameters[1] = {}) { return listPendingWorkspaceLandIntentsAsync(this, filter); } + resolveWorkspaceLandIntent(input: Parameters[1]) { return resolveWorkspaceLandIntentAsync(this, input); } + resolveOrphanedWorkspaceLandIntent(input: Parameters[1]) { return resolveOrphanedWorkspaceLandIntentAsync(this, input); } + /** FNXC:SymbolLock 2026-07-30-10:00: FN-8306 resolves only durable task declarations; PROMPT is never re-read here. */ async resolveTaskSymbols(taskId: string): Promise { try { diff --git a/packages/core/src/task-store/workspace-leases.ts b/packages/core/src/task-store/workspace-leases.ts new file mode 100644 index 0000000000..a9d7448c16 --- /dev/null +++ b/packages/core/src/task-store/workspace-leases.ts @@ -0,0 +1,261 @@ +import { and, eq, isNull, sql } from "drizzle-orm"; +import { projectOwnershipPartition } from "../postgres/data-layer.js"; +import type { DbTransaction } from "../postgres/data-layer.js"; +import * as schema from "../postgres/schema/index.js"; +import type { TaskStore } from "../store.js"; +import type { + AcquireWorkspaceLeaseResult, + WorkspaceLandIntent, + WorkspaceLandIntentResolution, + WorkspaceLease, + WorkspaceLeaseConflict, + WorkspaceLeaseHandle, + WorkspaceLeaseKind, + WorkspaceLeaseOwner, + WorkspaceLeaseReclaimOutcome, +} from "../tasks/workspace-lease-types.js"; +import { isTerminalWorkspaceLeaseOwner } from "../tasks/workspace-lease-types.js"; + +type LeaseRow = typeof schema.project.workspaceCoordinationLeases.$inferSelect; + +type LeaseStore = Pick; + +function project(store: LeaseStore) { + const layer = store.getAsyncLayer(); + if (!layer) throw new Error("Workspace leases require an AsyncDataLayer"); + return { layer, projectId: projectOwnershipPartition(layer.projectId) }; +} + +function assertLeaseInput(leaseKey: string, leaseMs?: number): void { + if (!leaseKey.trim()) throw new Error("Workspace lease key must not be empty"); + if (leaseMs !== undefined && (!Number.isFinite(leaseMs) || leaseMs <= 0)) { + throw new Error("Workspace leaseMs must be positive"); + } +} + +/** Serialize absent-row claims as well as updates to an existing lease row. */ +async function lock(tx: DbTransaction, projectId: string, leaseKey: string): Promise { + await tx.execute(sql`SELECT pg_advisory_xact_lock(hashtextextended(${`${projectId}:${leaseKey}`}, 0))`); +} + +function ownerMatches(row: Pick, owner: WorkspaceLeaseOwner): boolean { + return row.ownerTaskId === owner.taskId + && row.ownerNodeId === owner.nodeId + && row.ownerIncarnationId === owner.incarnationId + && (row.ownerRunId ?? undefined) === owner.runId; +} + +function handle(row: LeaseRow): WorkspaceLeaseHandle { + return { + leaseKey: row.leaseKey, + owner: { taskId: row.ownerTaskId, nodeId: row.ownerNodeId, incarnationId: row.ownerIncarnationId, ...(row.ownerRunId ? { runId: row.ownerRunId } : {}) }, + fenceToken: row.fenceToken, + expiresAt: row.expiresAt, + ...(row.fenceRefName ? { fenceRefName: row.fenceRefName } : {}), + ...(row.fenceRefSha ? { fenceRefSha: row.fenceRefSha } : {}), + }; +} + +function lease(row: LeaseRow): WorkspaceLease { + return { ...handle(row), kind: row.kind as WorkspaceLeaseKind, status: row.status as WorkspaceLease["status"], acquiredAt: row.acquiredAt, renewedAt: row.renewedAt }; +} + +function conflict(row: LeaseRow): WorkspaceLeaseConflict { + return { leaseKey: row.leaseKey, taskId: row.ownerTaskId, nodeId: row.ownerNodeId, incarnationId: row.ownerIncarnationId, fenceToken: row.fenceToken, expiresAt: row.expiresAt }; +} + +function intent(row: typeof schema.project.workspaceLandIntents.$inferSelect): WorkspaceLandIntent { + return { + taskId: row.taskId, repoRelPath: row.repoRelPath, remoteUrl: row.remoteUrl, integrationRef: row.integrationRef, + intendedSha: row.intendedSha, expectedTip: row.expectedTip, fenceRefName: row.fenceRefName, fenceRefSha: row.fenceRefSha, + owner: { taskId: row.ownerTaskId, nodeId: row.ownerNodeId, incarnationId: row.ownerIncarnationId }, fenceToken: row.fenceToken, + status: row.status as WorkspaceLandIntent["status"], ...(row.resolvedSha ? { resolvedSha: row.resolvedSha } : {}), + ...(row.resolution ? { resolution: row.resolution as WorkspaceLandIntentResolution } : {}), + createdAt: row.createdAt, updatedAt: row.updatedAt, ...(row.resolvedAt ? { resolvedAt: row.resolvedAt } : {}), + }; +} + +/** + * FNXC:WorkspaceLease 2026-08-15-08:45: + * Workspace acquire, land, and merge-dispatch writers contend on one durable + * project/repository key. The increasing token is the fencing authority; a TTL + * only permits a successor to claim it and never authorizes an old holder. + */ +export async function acquireWorkspaceLeaseAsync( + store: TaskStore, + input: { leaseKey: string; kind: WorkspaceLeaseKind; owner: WorkspaceLeaseOwner; leaseMs: number }, +): Promise { + assertLeaseInput(input.leaseKey, input.leaseMs); + const { layer, projectId } = project(store); + return layer.transactionImmediate(async (tx) => { + await lock(tx, projectId, input.leaseKey); + const [existing] = await tx.select().from(schema.project.workspaceCoordinationLeases).where(and( + eq(schema.project.workspaceCoordinationLeases.projectId, projectId), eq(schema.project.workspaceCoordinationLeases.leaseKey, input.leaseKey), + )).limit(1); + const now = new Date(); const nowIso = now.toISOString(); const expiresAt = new Date(now.getTime() + input.leaseMs).toISOString(); + if (existing?.status === "held" && existing.expiresAt > nowIso && !ownerMatches(existing, input.owner)) { + return { outcome: "conflict", conflict: conflict(existing) }; + } + if (existing?.status === "held" && existing.expiresAt > nowIso) { + if (existing.kind !== input.kind) return { outcome: "conflict", conflict: conflict(existing) }; + const [renewed] = await tx.update(schema.project.workspaceCoordinationLeases).set({ renewedAt: nowIso, expiresAt, updatedAt: nowIso }).where(and( + eq(schema.project.workspaceCoordinationLeases.projectId, projectId), eq(schema.project.workspaceCoordinationLeases.leaseKey, input.leaseKey), + eq(schema.project.workspaceCoordinationLeases.fenceToken, existing.fenceToken), eq(schema.project.workspaceCoordinationLeases.status, "held"), + )).returning(); + if (!renewed) throw new Error("Workspace lease reentrant renewal lost its fence"); + return { outcome: "reentrant", handle: handle(renewed) }; + } + const fenceToken = (existing?.fenceToken ?? 0n) + 1n; + const values = { + projectId, leaseKey: input.leaseKey, kind: input.kind, ownerTaskId: input.owner.taskId, ownerNodeId: input.owner.nodeId, + ownerIncarnationId: input.owner.incarnationId, ownerRunId: input.owner.runId ?? null, fenceToken, fenceRefName: null, fenceRefSha: null, + status: "held", acquiredAt: nowIso, renewedAt: nowIso, expiresAt, createdAt: existing?.createdAt ?? nowIso, updatedAt: nowIso, + }; + const [claimed] = existing + ? await tx.update(schema.project.workspaceCoordinationLeases).set(values).where(and(eq(schema.project.workspaceCoordinationLeases.projectId, projectId), eq(schema.project.workspaceCoordinationLeases.leaseKey, input.leaseKey))).returning() + : await tx.insert(schema.project.workspaceCoordinationLeases).values(values).returning(); + if (!claimed) throw new Error("Workspace lease claim did not return a row"); + /* + FNXC:WorkspaceLease 2026-08-15-12:00: + Only an absent row starts a new tenancy. Reusing released or expired rows + clears their pin and must be labelled reclaimed so fence publication occurs. + */ + return { outcome: existing ? "reclaimed-expired" : "acquired", handle: handle(claimed) }; + }); +} + +export async function renewWorkspaceLeaseAsync(store: TaskStore, current: WorkspaceLeaseHandle, leaseMs: number): Promise { + assertLeaseInput(current.leaseKey, leaseMs); + const { layer, projectId } = project(store); + return layer.transactionImmediate(async (tx) => { + await lock(tx, projectId, current.leaseKey); + const now = new Date(); const nowIso = now.toISOString(); const expiresAt = new Date(now.getTime() + leaseMs).toISOString(); + const [renewed] = await tx.update(schema.project.workspaceCoordinationLeases).set({ renewedAt: nowIso, expiresAt, updatedAt: nowIso }).where(and( + eq(schema.project.workspaceCoordinationLeases.projectId, projectId), eq(schema.project.workspaceCoordinationLeases.leaseKey, current.leaseKey), + eq(schema.project.workspaceCoordinationLeases.ownerTaskId, current.owner.taskId), eq(schema.project.workspaceCoordinationLeases.ownerNodeId, current.owner.nodeId), + eq(schema.project.workspaceCoordinationLeases.ownerIncarnationId, current.owner.incarnationId), (current.owner.runId ? eq(schema.project.workspaceCoordinationLeases.ownerRunId, current.owner.runId) : isNull(schema.project.workspaceCoordinationLeases.ownerRunId)), + eq(schema.project.workspaceCoordinationLeases.fenceToken, current.fenceToken), eq(schema.project.workspaceCoordinationLeases.status, "held"), + )).returning(); + return renewed ? handle(renewed) : undefined; + }); +} + +export async function releaseWorkspaceLeaseAsync(store: TaskStore, current: WorkspaceLeaseHandle): Promise { + const { layer, projectId } = project(store); + return layer.transactionImmediate(async (tx) => { + await lock(tx, projectId, current.leaseKey); + const released = await tx.update(schema.project.workspaceCoordinationLeases).set({ status: "released", updatedAt: new Date().toISOString() }).where(and( + eq(schema.project.workspaceCoordinationLeases.projectId, projectId), eq(schema.project.workspaceCoordinationLeases.leaseKey, current.leaseKey), + eq(schema.project.workspaceCoordinationLeases.ownerTaskId, current.owner.taskId), eq(schema.project.workspaceCoordinationLeases.ownerNodeId, current.owner.nodeId), + eq(schema.project.workspaceCoordinationLeases.ownerIncarnationId, current.owner.incarnationId), (current.owner.runId ? eq(schema.project.workspaceCoordinationLeases.ownerRunId, current.owner.runId) : isNull(schema.project.workspaceCoordinationLeases.ownerRunId)), eq(schema.project.workspaceCoordinationLeases.fenceToken, current.fenceToken), eq(schema.project.workspaceCoordinationLeases.status, "held"), + )).returning({ leaseKey: schema.project.workspaceCoordinationLeases.leaseKey }); + return released.length === 1; + }); +} + +export async function withValidWorkspaceLeaseAsync(store: TaskStore, current: WorkspaceLeaseHandle, fn: (tx: DbTransaction) => Promise): Promise { + const { layer, projectId } = project(store); + return layer.transactionImmediate(async (tx) => { + await lock(tx, projectId, current.leaseKey); + const [row] = await tx.select().from(schema.project.workspaceCoordinationLeases).where(and(eq(schema.project.workspaceCoordinationLeases.projectId, projectId), eq(schema.project.workspaceCoordinationLeases.leaseKey, current.leaseKey))).limit(1); + if (!row || row.status !== "held" || row.expiresAt <= new Date().toISOString() || !ownerMatches(row, current.owner) || row.fenceToken !== current.fenceToken) { + throw new Error("Workspace lease is no longer valid"); + } + return fn(tx); + }); +} + +export async function validateWorkspaceLeaseFenceAsync(store: TaskStore, input: { leaseKey: string; owner: WorkspaceLeaseOwner; fenceToken: bigint }) { + const { layer, projectId } = project(store); + const [row] = await layer.db.select().from(schema.project.workspaceCoordinationLeases).where(and(eq(schema.project.workspaceCoordinationLeases.projectId, projectId), eq(schema.project.workspaceCoordinationLeases.leaseKey, input.leaseKey))).limit(1); + if (!row) return "missing" as const; + if (row.fenceToken < input.fenceToken) return "unknown" as const; + if (row.fenceToken !== input.fenceToken || !ownerMatches(row, input.owner) || row.status !== "held" || row.expiresAt <= new Date().toISOString()) return "superseded" as const; + return "valid" as const; +} + +export async function recordWorkspaceLeaseFenceRefAsync(store: TaskStore, input: { handle: WorkspaceLeaseHandle; fenceRefName: string; fenceRefSha: string }): Promise { + if (!input.fenceRefName || !input.fenceRefSha) throw new Error("Workspace fence ref name and SHA are required"); + return withValidWorkspaceLeaseAsync(store, input.handle, async (tx) => { + const { projectId } = project(store); + const [updated] = await tx.update(schema.project.workspaceCoordinationLeases).set({ fenceRefName: input.fenceRefName, fenceRefSha: input.fenceRefSha, updatedAt: new Date().toISOString() }).where(and( + eq(schema.project.workspaceCoordinationLeases.projectId, projectId), eq(schema.project.workspaceCoordinationLeases.leaseKey, input.handle.leaseKey), eq(schema.project.workspaceCoordinationLeases.fenceToken, input.handle.fenceToken), + )).returning(); + if (!updated) throw new Error("Workspace fence ref write lost its lease"); + return handle(updated); + }); +} + +export async function inspectWorkspaceLeasesAsync(store: TaskStore, filter: { taskId?: string; leaseKeys?: string[] } = {}): Promise { + const { layer, projectId } = project(store); + const conditions = [eq(schema.project.workspaceCoordinationLeases.projectId, projectId)]; + if (filter.taskId) conditions.push(eq(schema.project.workspaceCoordinationLeases.ownerTaskId, filter.taskId)); + const rows = await layer.db.select().from(schema.project.workspaceCoordinationLeases).where(and(...conditions)); + return rows.filter((row) => !filter.leaseKeys || filter.leaseKeys.includes(row.leaseKey)).map(lease); +} + +export async function recordWorkspaceLandIntentAsync(store: TaskStore, input: { handle: WorkspaceLeaseHandle; taskId: string; repoRelPath: string; remoteUrl: string; integrationRef: string; intendedSha: string; expectedTip: string }) { + if (!input.handle.fenceRefName || !input.handle.fenceRefSha) throw new Error("Land intent requires a published fence ref"); + const fenceRefName = input.handle.fenceRefName; + const fenceRefSha = input.handle.fenceRefSha; + return withValidWorkspaceLeaseAsync(store, input.handle, async (tx) => { + const { projectId } = project(store); const now = new Date().toISOString(); + const values: typeof schema.project.workspaceLandIntents.$inferInsert = { projectId, taskId: input.taskId, repoRelPath: input.repoRelPath, remoteUrl: input.remoteUrl, integrationRef: input.integrationRef, intendedSha: input.intendedSha, expectedTip: input.expectedTip, fenceRefName, fenceRefSha, ownerTaskId: input.handle.owner.taskId, ownerNodeId: input.handle.owner.nodeId, ownerIncarnationId: input.handle.owner.incarnationId, fenceToken: input.handle.fenceToken, status: "pending", createdAt: now, updatedAt: now }; + await tx.insert(schema.project.workspaceLandIntents).values(values).onConflictDoUpdate({ target: [schema.project.workspaceLandIntents.projectId, schema.project.workspaceLandIntents.taskId, schema.project.workspaceLandIntents.repoRelPath], set: { remoteUrl: input.remoteUrl, integrationRef: input.integrationRef, intendedSha: input.intendedSha, expectedTip: input.expectedTip, fenceRefName, fenceRefSha, ownerTaskId: input.handle.owner.taskId, ownerNodeId: input.handle.owner.nodeId, ownerIncarnationId: input.handle.owner.incarnationId, fenceToken: input.handle.fenceToken, status: "pending", resolvedSha: null, resolution: null, resolvedAt: null, updatedAt: now } }); + }); +} + +export async function listPendingWorkspaceLandIntentsAsync(store: TaskStore, filter: { taskId?: string; repoRelPath?: string; limit?: number } = {}): Promise { + const { layer, projectId } = project(store); const conditions = [eq(schema.project.workspaceLandIntents.projectId, projectId), eq(schema.project.workspaceLandIntents.status, "pending")]; + if (filter.taskId) conditions.push(eq(schema.project.workspaceLandIntents.taskId, filter.taskId)); + if (filter.repoRelPath) conditions.push(eq(schema.project.workspaceLandIntents.repoRelPath, filter.repoRelPath)); + const rows = await layer.db.select().from(schema.project.workspaceLandIntents).where(and(...conditions)); + return rows.slice(0, filter.limit).map(intent); +} + +async function resolveIntent(tx: DbTransaction, projectId: string, input: { taskId: string; repoRelPath: string; expectedIntentFenceToken: bigint; resolution: WorkspaceLandIntentResolution; resolvedSha?: string; persistLandedSha?: () => Promise }, maxFence?: bigint) { + const [row] = await tx.select().from(schema.project.workspaceLandIntents).where(and(eq(schema.project.workspaceLandIntents.projectId, projectId), eq(schema.project.workspaceLandIntents.taskId, input.taskId), eq(schema.project.workspaceLandIntents.repoRelPath, input.repoRelPath))).limit(1); + if (!row) return { outcome: "missing" as const }; + if (row.status !== "pending") return { outcome: "resolved" as const }; + if (row.fenceToken !== input.expectedIntentFenceToken || (maxFence !== undefined && row.fenceToken > maxFence)) return { outcome: "stale-intent" as const }; + if (input.resolution === "landed") await input.persistLandedSha?.(); + const now = new Date().toISOString(); + await tx.update(schema.project.workspaceLandIntents).set({ status: input.resolution === "landed" ? "recorded" : "abandoned", resolution: input.resolution, resolvedSha: input.resolvedSha ?? null, resolvedAt: now, updatedAt: now }).where(and(eq(schema.project.workspaceLandIntents.projectId, projectId), eq(schema.project.workspaceLandIntents.taskId, input.taskId), eq(schema.project.workspaceLandIntents.repoRelPath, input.repoRelPath), eq(schema.project.workspaceLandIntents.status, "pending"))); + return { outcome: "resolved" as const }; +} + +export async function resolveWorkspaceLandIntentAsync(store: TaskStore, input: { handle: WorkspaceLeaseHandle; taskId: string; repoRelPath: string; expectedIntentFenceToken: bigint; resolution: WorkspaceLandIntentResolution; resolvedSha?: string; persistLandedSha?: () => Promise }) { + return withValidWorkspaceLeaseAsync(store, input.handle, (tx) => resolveIntent(tx, project(store).projectId, input, input.handle.fenceToken)); +} + +export async function resolveOrphanedWorkspaceLandIntentAsync(store: TaskStore, input: { leaseKey: string; taskId: string; repoRelPath: string; expectedIntentFenceToken: bigint; resolution: WorkspaceLandIntentResolution; resolvedSha?: string; persistLandedSha?: () => Promise }) { + const { layer, projectId } = project(store); + return layer.transactionImmediate(async (tx) => { await lock(tx, projectId, input.leaseKey); const [row] = await tx.select().from(schema.project.workspaceCoordinationLeases).where(and(eq(schema.project.workspaceCoordinationLeases.projectId, projectId), eq(schema.project.workspaceCoordinationLeases.leaseKey, input.leaseKey))).limit(1); if (row?.status === "held" && row.expiresAt > new Date().toISOString()) return { outcome: "lease-live" as const }; return resolveIntent(tx, projectId, input); }); +} + +export async function reclaimWorkspaceLeaseAsync(store: TaskStore, input: { leaseKey: string; expectedOwner: WorkspaceLeaseOwner; expectedFenceToken: bigint; requireTerminalOwner?: boolean; reason?: string }): Promise<{ outcome: WorkspaceLeaseReclaimOutcome }> { + const { layer, projectId } = project(store); + return layer.transactionImmediate(async (tx) => { + await lock(tx, projectId, input.leaseKey); const [row] = await tx.select().from(schema.project.workspaceCoordinationLeases).where(and(eq(schema.project.workspaceCoordinationLeases.projectId, projectId), eq(schema.project.workspaceCoordinationLeases.leaseKey, input.leaseKey))).limit(1); + if (!row) return { outcome: "missing" }; if (!ownerMatches(row, input.expectedOwner) || row.fenceToken !== input.expectedFenceToken) return { outcome: "stale-precondition" }; + if (row.expiresAt > new Date().toISOString()) { + if (!input.requireTerminalOwner) return { outcome: "still-live" }; + const [owner] = await tx.select({ column: schema.project.tasks.column, status: schema.project.tasks.status }).from(schema.project.tasks).where(and(eq(schema.project.tasks.projectId, projectId), eq(schema.project.tasks.id, row.ownerTaskId))).limit(1); + if (!owner) return { outcome: "owner-unresolvable" }; if (!isTerminalWorkspaceLeaseOwner({ ...owner, status: owner.status ?? undefined })) return { outcome: "still-live" }; + } + await tx.update(schema.project.workspaceCoordinationLeases).set({ status: "expired", updatedAt: new Date().toISOString() }).where(and(eq(schema.project.workspaceCoordinationLeases.projectId, projectId), eq(schema.project.workspaceCoordinationLeases.leaseKey, input.leaseKey), eq(schema.project.workspaceCoordinationLeases.fenceToken, input.expectedFenceToken), eq(schema.project.workspaceCoordinationLeases.status, "held"))); + return { outcome: "reclaimed" }; + }); +} + +export async function reconcileExpiredWorkspaceLeasesAsync(store: TaskStore): Promise { + const leases = await inspectWorkspaceLeasesAsync(store); const reclaimed: string[] = []; + for (const item of leases) if (item.status === "held" && item.expiresAt <= new Date().toISOString()) { const result = await reclaimWorkspaceLeaseAsync(store, { leaseKey: item.leaseKey, expectedOwner: item.owner, expectedFenceToken: item.fenceToken }); if (result.outcome === "reclaimed") reclaimed.push(item.leaseKey); } + return reclaimed; +} + +export async function releaseStaleWorkspaceLeasesForNodeAsync(store: TaskStore, nodeId: string, { currentIncarnationId }: { currentIncarnationId: string }): Promise { + const leases = await inspectWorkspaceLeasesAsync(store); const reclaimed: string[] = []; + for (const item of leases) if (item.owner.nodeId === nodeId && item.owner.incarnationId !== currentIncarnationId && item.expiresAt <= new Date().toISOString()) { const result = await reclaimWorkspaceLeaseAsync(store, { leaseKey: item.leaseKey, expectedOwner: item.owner, expectedFenceToken: item.fenceToken }); if (result.outcome === "reclaimed") reclaimed.push(item.leaseKey); } + return reclaimed; +} diff --git a/packages/core/src/tasks/workspace-lease-types.ts b/packages/core/src/tasks/workspace-lease-types.ts new file mode 100644 index 0000000000..69539cc3fe --- /dev/null +++ b/packages/core/src/tasks/workspace-lease-types.ts @@ -0,0 +1,24 @@ +import type { Task } from "../types.js"; + +export type WorkspaceLeaseKind = "acquire" | "land" | "merge-dispatch"; +export type WorkspaceLeaseStatus = "held" | "released" | "expired"; +export type WorkspaceLeaseOwner = { taskId: string; nodeId: string; incarnationId: string; runId?: string }; +export type WorkspaceLeaseHandle = { leaseKey: string; owner: WorkspaceLeaseOwner; fenceToken: bigint; expiresAt: string; fenceRefName?: string; fenceRefSha?: string }; +export type WorkspaceLeaseClaimOutcome = "acquired" | "reentrant" | "reclaimed-expired" | "conflict"; +export type WorkspaceLeaseFenceOutcome = "valid" | "superseded" | "missing" | "unknown"; +export type WorkspaceLandIntentResolution = "landed" | "not-landed"; +export type WorkspaceLandIntentResolveOutcome = "resolved" | "lease-live" | "stale-intent" | "superseded" | "missing"; +export type WorkspaceLeaseReclaimOutcome = "reclaimed" | "stale-precondition" | "still-live" | "owner-unresolvable" | "missing"; +export interface WorkspaceLease extends WorkspaceLeaseHandle { kind: WorkspaceLeaseKind; status: WorkspaceLeaseStatus; acquiredAt: string; renewedAt: string; } +export interface WorkspaceLeaseConflict { leaseKey: string; taskId: string; nodeId: string; incarnationId: string; fenceToken: bigint; expiresAt: string; } +export type AcquireWorkspaceLeaseResult = { outcome: Exclude; handle: WorkspaceLeaseHandle } | { outcome: "conflict"; conflict: WorkspaceLeaseConflict }; +export interface WorkspaceLandIntent { taskId: string; repoRelPath: string; remoteUrl: string; integrationRef: string; intendedSha: string; expectedTip: string; fenceRefName: string; fenceRefSha: string; owner: WorkspaceLeaseOwner; fenceToken: bigint; status: "pending" | "recorded" | "abandoned"; resolvedSha?: string; resolution?: WorkspaceLandIntentResolution; createdAt: string; updatedAt: string; resolvedAt?: string; } + +/** + * FNXC:Workspace 2026-08-15-08:23: + * Store reclaim and workspace self-healing share this deliberately narrow + * terminal rule so either cannot reclaim a task the other considers live. + */ +export function isTerminalWorkspaceLeaseOwner(row: Pick | null | undefined): boolean { + return row != null && (row.column === "done" || row.status === "failed"); +} diff --git a/packages/engine/src/__tests__/project-engine.test.ts b/packages/engine/src/__tests__/project-engine.test.ts index dc74ebb79b..f35c0095cf 100644 --- a/packages/engine/src/__tests__/project-engine.test.ts +++ b/packages/engine/src/__tests__/project-engine.test.ts @@ -1697,6 +1697,71 @@ describe("ProjectEngine workspace merge dispatch hardening (Phase C review)", () ...overrides, }); + /* + FNXC:WorkspaceMergeDispatch 2026-08-15-08:56: + The queue's process-local Set cannot serialize two ProjectEngine instances. These + dispatch tests use the real queue path and only mock the durable lease boundary. + */ + it("releases its durable workspace dispatch claim after a successful manual land", async () => { + const mockStore = createMockStore({ ...baseSettings, autoMerge: true }); + const task = workspaceTask(); + mockStore.store.getTask.mockResolvedValue(task); + const handle = { + leaseKey: "merge-dispatch:FN-WSH", + owner: { taskId: "FN-WSH", nodeId: "node-a", incarnationId: "run-a" }, + fenceToken: 1n, + expiresAt: "2026-08-15T09:01:00.000Z", + }; + const acquireWorkspaceLease = vi.fn(async () => ({ outcome: "acquired" as const, handle })); + const releaseWorkspaceLease = vi.fn(async () => true); + Object.assign(mockStore.store, { acquireWorkspaceLease, releaseWorkspaceLease }); + mocks.currentStore = mockStore.store; + mocks.landWorkspaceTask.mockResolvedValue({ + allLanded: true, + finalized: true, + repos: [{ repo: "repo-a", status: "landed", landedSha: "aaaa1111", integrationBranch: "main" }], + }); + + const engine = createEngine(); + await engine.start(); + await engine.onMerge("FN-WSH"); + + expect(acquireWorkspaceLease).toHaveBeenCalledWith(expect.objectContaining({ + leaseKey: "merge-dispatch:FN-WSH", + kind: "merge-dispatch", + owner: expect.objectContaining({ taskId: "FN-WSH" }), + })); + expect(releaseWorkspaceLease).toHaveBeenCalledWith(handle); + await engine.stop(); + }); + + it("fails a conflicting manual dispatch without calling the workspace land body", async () => { + const mockStore = createMockStore({ ...baseSettings, autoMerge: true }); + mockStore.store.getTask.mockResolvedValue(workspaceTask()); + const acquireWorkspaceLease = vi.fn(async () => ({ + outcome: "conflict" as const, + conflict: { + leaseKey: "merge-dispatch:FN-WSH", + taskId: "FN-other", + nodeId: "node-b", + incarnationId: "run-b", + fenceToken: 2n, + expiresAt: "2026-08-15T09:01:00.000Z", + }, + })); + Object.assign(mockStore.store, { acquireWorkspaceLease }); + mocks.currentStore = mockStore.store; + + const engine = createEngine(); + await engine.start(); + await expect(engine.onMerge("FN-WSH")).rejects.toThrow("workspace merge dispatch is in progress for task FN-other"); + + expect(acquireWorkspaceLease).toHaveBeenCalledTimes(1); + expect(mocks.landWorkspaceTask).not.toHaveBeenCalled(); + expect(mockStore.store.updateTask).not.toHaveBeenCalled(); + await engine.stop(); + }); + it("rejects a manual workspace merge when finalization is blocked without laundering success", async () => { const mockStore = createMockStore({ ...baseSettings, autoMerge: true }); const task = workspaceTask({ @@ -2665,7 +2730,7 @@ describe("ProjectEngine paused in-review auto-merge behavior", () => { }; expect(privateEngine.mergeQueue).not.toContain("FN-MERGE-WAITING"); expect(privateEngine.capacityDeferredMergeTaskIds.has("FN-MERGE-WAITING")).toBe(true); - expect(engine.isMergePending("FN-MERGE-WAITING")).toBe(true); + expect(await engine.isMergePending("FN-MERGE-WAITING")).toBe(true); expect(mockStore.store.logEntry).toHaveBeenCalledWith( "FN-MERGE-WAITING", expect.stringContaining("maxWorktrees capacity exhausted: used=1/1"), @@ -2689,7 +2754,7 @@ describe("ProjectEngine paused in-review auto-merge behavior", () => { expect(mocks.runAiMerge).not.toHaveBeenCalled(); await engine.stop(); expect(privateEngine.capacityDeferredMergeTaskIds.size).toBe(0); - expect(engine.isMergePending("FN-MERGE-WAITING")).toBe(false); + expect(await engine.isMergePending("FN-MERGE-WAITING")).toBe(false); }); it("records an audit event (not silent) when auto-promotion of a branch-group member fails (Fix #4)", async () => { diff --git a/packages/engine/src/__tests__/self-healing-workspace.test.ts b/packages/engine/src/__tests__/self-healing-workspace.test.ts index 42290f43d1..a6b6db737c 100644 --- a/packages/engine/src/__tests__/self-healing-workspace.test.ts +++ b/packages/engine/src/__tests__/self-healing-workspace.test.ts @@ -22,7 +22,7 @@ import { EventEmitter } from "node:events"; import { execSync } from "node:child_process"; import { existsSync, mkdirSync, realpathSync, rmSync, symlinkSync, writeFileSync } from "node:fs"; import path from "node:path"; -import { registerArchiveWorkspaceWorktreeDisposer, type Settings, type Task, type TaskStore } from "@fusion/core"; +import { registerArchiveWorkspaceWorktreeDisposer, type Settings, type Task, type TaskStore, type WorkspaceLandIntent } from "@fusion/core"; import { createSharedPgTaskStoreTestHarness, pgDescribe, type SharedPgTaskStoreHarness } from "../../../core/src/__test-utils__/pg-test-harness.js"; import { SelfHealingManager } from "../self-healing.js"; import { classifyBranchProbeError } from "../self-healing-git-evidence.js"; @@ -154,6 +154,20 @@ class PruneFailureWorkspaceTeardownManager extends SelfHealingManager { } } +class WorkspaceLandIntentManager extends SelfHealingManager { + constructor( + store: TaskStore, + options: ConstructorParameters[1], + private readonly evidence: { resolution: "landed"; resolvedSha: string } | { resolution: "not-landed" } | undefined, + ) { + super(store, options); + } + + protected override async readWorkspaceLandIntentRemoteEvidence(_intent: WorkspaceLandIntent): Promise<{ resolution: "landed"; resolvedSha: string } | { resolution: "not-landed" } | undefined> { + return this.evidence; + } +} + /** Add a real `fusion/` branch in a sub-repo with one non-conflicting own commit. */ function addRepoBranch(fx: WorkspaceFixture, repoRel: string, content: string): void { const repoDir = fx.repoPath(repoRel); @@ -292,6 +306,53 @@ describeIfGit("workspace-aware self-healing (Phase D U1)", () => { fx?.cleanup(); }); + /* + FNXC:Workspace 2026-08-15-08:59: + A different engine node must be able to turn an interrupted fenced push into the exact + per-repository landed SHA. The resolver callback is the durable seam: it persists first and + records the intent only after that task-map update succeeds. + */ + it("reconciles a remotely landed pending intent without the original node", async () => { + fx = await createWorkspaceFixture(["repo-a"]); + const task = workspaceTask({ "repo-a": { worktreePath: fx.repoPath("repo-a"), branch: BRANCH } }); + const store = createStore([task]); + const intent: WorkspaceLandIntent = { + taskId: TASK_ID, + repoRelPath: "repo-a", + remoteUrl: "https://example.test/repo-a.git", + integrationRef: "refs/heads/main", + intendedSha: "landed-sha", + expectedTip: "prior-sha", + fenceRefName: "refs/fusion/fence/repo-a", + fenceRefSha: "fence-sha", + owner: { taskId: TASK_ID, nodeId: "dead-node", incarnationId: "dead-incarnation" }, + fenceToken: 7n, + status: "pending", + createdAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + }; + const listPendingWorkspaceLandIntents = vi.fn().mockResolvedValue([intent]); + const resolveOrphanedWorkspaceLandIntent = vi.fn(async (input: { persistLandedSha?: () => Promise }) => { + await input.persistLandedSha?.(); + return { outcome: "resolved" }; + }); + Object.assign(store, { listPendingWorkspaceLandIntents, resolveOrphanedWorkspaceLandIntent }); + const manager = new WorkspaceLandIntentManager( + store, + managerOptions(store, fx.rootDir) as never, + { resolution: "landed", resolvedSha: "landed-sha" }, + ); + + expect(await manager.reconcilePendingWorkspaceLandIntents()).toBe(1); + expect(resolveOrphanedWorkspaceLandIntent).toHaveBeenCalledWith(expect.objectContaining({ + leaseKey: "repo:repo-a", + expectedIntentFenceToken: 7n, + resolution: "landed", + resolvedSha: "landed-sha", + })); + expect(store.tasks.get(TASK_ID)?.workspaceWorktrees?.["repo-a"]?.landedSha).toBe("landed-sha"); + }); + // ── KTD1 P0: partial-landed "merging" task must NOT be finalized done ────── it("recoverInterruptedMergingTasks does NOT finalize a partial-landed workspace task (P0)", async () => { fx = await createWorkspaceFixture(["repo-a", "repo-b"]); diff --git a/packages/engine/src/__tests__/workspace-durable-coordination.test.ts b/packages/engine/src/__tests__/workspace-durable-coordination.test.ts new file mode 100644 index 0000000000..559174a448 --- /dev/null +++ b/packages/engine/src/__tests__/workspace-durable-coordination.test.ts @@ -0,0 +1,49 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { WorkspaceLeaseHandle } from "@fusion/core"; +import { ensureTenancyFenceRef, mergeDispatchFenceRef } from "../merge/workspace-fence-ref.js"; +import { createWorkspaceFixture, hasGit, type WorkspaceFixture } from "./_workspace-fixture.js"; + +const describeIfGit = hasGit ? describe : describe.skip; + +/* +FNXC:WorkspaceMergeDispatch 2026-08-15-09:46: +A dispatch tenancy is global to a workspace task while its protected resources live in separate +sub-repo remotes. Reentry must retain one durable pin and publish that same pin to each remote, +not assume a workspace root checkout has an origin that can fence all subsequent land pushes. +*/ +describeIfGit("workspace durable dispatch coordination", () => { + let fx: WorkspaceFixture; + afterEach(() => fx?.cleanup()); + + it("publishes a reentrant dispatch pin to a second remote without rotating the tenancy", async () => { + fx = await createWorkspaceFixture(["repo-a", "repo-b"]); + for (const repoRel of fx.repos) { + const remote = `${fx.rootDir}/${repoRel}.git`; + fx.git(repoRel, `git init --bare ${remote}`); + fx.git(repoRel, `git remote add origin ${remote}`); + fx.git(repoRel, "git push -u origin main"); + } + const initial: WorkspaceLeaseHandle = { + leaseKey: "merge-dispatch:FN-9059", kind: "merge-dispatch", + owner: { taskId: "FN-9059", nodeId: "node-a", incarnationId: "inc-a" }, fenceToken: 7n, + }; + const record = vi.fn(async (input: { handle: WorkspaceLeaseHandle; fenceRefName: string; fenceRefSha: string }) => ({ + ...input.handle, fenceRefName: input.fenceRefName, fenceRefSha: input.fenceRefSha, + })); + const fenceRefName = mergeDispatchFenceRef("FN-9059"); + const first = await ensureTenancyFenceRef({ + store: { recordWorkspaceLeaseFenceRef: record }, handle: initial, claimOutcome: "acquired", + remote: "origin", cwd: fx.repoPath("repo-a"), fenceRefName, + }); + const reentrant = await ensureTenancyFenceRef({ + store: { recordWorkspaceLeaseFenceRef: record }, handle: first, claimOutcome: "reentrant", + remote: "origin", cwd: fx.repoPath("repo-b"), fenceRefName, + }); + + expect(record).toHaveBeenCalledOnce(); + expect(reentrant).toEqual(first); + for (const repoRel of fx.repos) { + expect(fx.git(repoRel, `git ls-remote origin ${fenceRefName}`).split(/\s+/)[0]).toBe(first.fenceRefSha); + } + }); +}); diff --git a/packages/engine/src/__tests__/workspace-merger-lease.test.ts b/packages/engine/src/__tests__/workspace-merger-lease.test.ts index 6fd3be4d1e..5655fb6655 100644 --- a/packages/engine/src/__tests__/workspace-merger-lease.test.ts +++ b/packages/engine/src/__tests__/workspace-merger-lease.test.ts @@ -21,13 +21,15 @@ Coverage (FN-5893 surfaces): - 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 { afterAll, afterEach, beforeAll, 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 "../merge/merger-ai.js"; +import { createSharedPgTaskStoreTestHarness, pgDescribe, type SharedPgTaskStoreHarness } from "../../../core/src/__test-utils__/pg-test-harness.js"; +import { landSquash, landWorkspaceTask, WorkspaceMergeDispatchSupersededError, WorkspaceRepoLandBusyError } from "../merge/merger-ai.js"; +import { ensureTenancyFenceRef, mergeDispatchFenceRef, WorkspaceFenceRefError } from "../merge/workspace-fence-ref.js"; import { activeSessionRegistry } from "../agents/active-session-registry.js"; import { createWorkspaceFixture, hasGit, type WorkspaceFixture } from "./_workspace-fixture.js"; @@ -133,6 +135,175 @@ function makeTask(id: string, workspaceWorktrees: Task["workspaceWorktrees"]): T } as Task; } +const pgDescribeIfGit = hasGit ? pgDescribe : describe.skip; + +/* +FNXC:WorkspaceMergeDispatch 2026-08-15-10:09: +The post-push finalizer must be proven through the production workspace land path and a real +PostgreSQL owner-and-fence reclaim. A mocked guarded writer cannot show that landWorkspaceTask +reaches the guarded terminal mutation after git has advanced the integration ref. +*/ +pgDescribeIfGit("workspace land dispatch finalization (PostgreSQL)", () => { + const h: SharedPgTaskStoreHarness = createSharedPgTaskStoreTestHarness({ + prefix: "fusion_workspace_land_dispatch_finalization", + }); + let fx: WorkspaceFixture; + + beforeAll(h.beforeAll); + beforeEach(async () => { + await h.beforeEach(); + fx = await createWorkspaceFixture(["repo-a"]); + }); + afterEach(async () => { + fx?.cleanup(); + await h.afterEach(); + }); + afterAll(h.afterAll); + + it("rejects a predecessor's real repo-b push after a successor republished its dispatch fence", async () => { + fx.cleanup(); + fx = await createWorkspaceFixture(["repo-a", "repo-b"]); + const store = h.store(); + const taskId = "FN-9059-PG-REPO-B"; + for (const repoRel of fx.repos) { + const remote = path.join(fx.rootDir, `${repoRel}.git`); + execSync(`git init --bare ${remote}`, { stdio: "pipe" }); + fx.git(repoRel, `git remote add origin ${remote}`); + fx.git(repoRel, "git push -u origin main"); + addRepoBranchWithEdit(fx, repoRel, taskId, `${repoRel} stale dispatch push\n`); + } + await store.createTaskWithReservedId( + { description: "cross-node repo-b dispatch fence", column: "in-review" }, + { taskId, applyDefaultWorkflowSteps: false }, + ); + await store.updateTask(taskId, { + branch: BRANCH, + workspaceWorktrees: Object.fromEntries(fx.repos.map((repoRel) => [repoRel, { + worktreePath: fx.repoPath(repoRel), branch: BRANCH, + }])), + } as Partial); + const task = (await store.getTask(taskId))!; + const predecessor = await store.acquireWorkspaceLease({ + leaseKey: `merge-dispatch:${taskId}`, + kind: "merge-dispatch", + owner: { taskId, nodeId: "node-b", incarnationId: "inc-b" }, + leaseMs: 60_000, + }); + if (predecessor.outcome === "conflict") throw new Error("expected predecessor dispatch lease"); + const repoBTip = fx.git("repo-b", "git rev-parse refs/heads/main"); + let successorClaimed = false; + let mergeAgentCalls = 0; + + await expect(landWorkspaceTask(store, task, fx.rootDir, { + workspaceDispatchFence: predecessor.handle, + }, { + mergeAgent: squashMergeAgent(BRANCH, async () => { + mergeAgentCalls++; + if (mergeAgentCalls !== 2) return; + await h.adminSql().unsafe(`UPDATE project.workspace_coordination_leases + SET expires_at = '${new Date(Date.now() - 1_000).toISOString()}' + WHERE lease_key = '${predecessor.handle.leaseKey}'`); + const successor = await store.acquireWorkspaceLease({ + leaseKey: predecessor.handle.leaseKey, + kind: "merge-dispatch", + owner: { taskId, nodeId: "node-a", incarnationId: "inc-a" }, + leaseMs: 60_000, + }); + expect(successor.outcome).toBe("reclaimed-expired"); + let successorHandle = successor.handle; + for (const repoRel of fx.repos) { + successorHandle = await ensureTenancyFenceRef({ + store, + handle: successorHandle, + claimOutcome: successor.outcome, + remote: "origin", + cwd: fx.repoPath(repoRel), + fenceRefName: mergeDispatchFenceRef(taskId), + }); + } + successorClaimed = true; + }), + reviewAgent: approveReviewAgent, + })).rejects.toBeInstanceOf(WorkspaceRepoLandBusyError); + + expect(successorClaimed).toBe(true); + // FNXC:WorkspaceMergeDispatch 2026-08-15-10:09: repo-b's target was unchanged; git refused only its stale dispatch ref pin. + expect(fx.git("repo-b", "git rev-parse refs/heads/main")).toBe(repoBTip); + expect(fx.git("repo-b", `git ls-remote origin ${mergeDispatchFenceRef(taskId)}`)).toMatch(/^[0-9a-f]{40,64}\s/); + }); + + it("leaves a real pushed land unfinalized when a successor reclaims the dispatch fence", async () => { + const store = h.store(); + const taskId = "FN-9059-PG-DISPATCH"; + const repo = fx.repoPath("repo-a"); + const remote = path.join(fx.rootDir, "origin.git"); + execSync(`git init --bare ${remote}`, { stdio: "pipe" }); + fx.git("repo-a", `git remote add origin ${remote}`); + fx.git("repo-a", "git push -u origin main"); + addRepoBranchWithEdit(fx, "repo-a", taskId, "production finalization fence\n"); + + await store.createTaskWithReservedId( + { description: "production workspace dispatch finalization", column: "in-review" }, + { taskId, applyDefaultWorkflowSteps: false }, + ); + await store.updateTask(taskId, { + branch: BRANCH, + workspaceWorktrees: { "repo-a": { worktreePath: repo, branch: BRANCH } }, + } as Partial); + const task = (await store.getTask(taskId))!; + const tipBefore = fx.git("repo-a", "git rev-parse refs/heads/main"); + const predecessor = await store.acquireWorkspaceLease({ + leaseKey: `merge-dispatch:${taskId}`, + kind: "merge-dispatch", + owner: { taskId, nodeId: "node-b", incarnationId: "inc-b" }, + leaseMs: 60_000, + }); + if (predecessor.outcome === "conflict") throw new Error("expected predecessor dispatch lease"); + + const mergeWorkspaceWorktreeEntry = store.mergeWorkspaceWorktreeEntry.bind(store); + let successorClaimed = false; + let terminalCallbackEntered = false; + const realWithValidWorkspaceLease = store.withValidWorkspaceLease.bind(store); + const finalizer = vi.spyOn(store, "withValidWorkspaceLease").mockImplementation(async (handle, callback) => + realWithValidWorkspaceLease(handle, async () => { + terminalCallbackEntered = true; + return callback(); + })); + vi.spyOn(store, "mergeWorkspaceWorktreeEntry").mockImplementation(async (...args) => { + const result = await mergeWorkspaceWorktreeEntry(...args); + if (!successorClaimed && args[0] === taskId && args[1] === "repo-a") { + successorClaimed = true; + await h.adminSql().unsafe(`UPDATE project.workspace_coordination_leases + SET expires_at = '${new Date(Date.now() - 1_000).toISOString()}' + WHERE lease_key = '${predecessor.handle.leaseKey}'`); + const successor = await store.acquireWorkspaceLease({ + leaseKey: predecessor.handle.leaseKey, + kind: "merge-dispatch", + owner: { taskId, nodeId: "node-a", incarnationId: "inc-a" }, + leaseMs: 60_000, + }); + expect(successor.outcome).toBe("reclaimed-expired"); + } + return result; + }); + + await expect(landWorkspaceTask(store, task, fx.rootDir, { + workspaceDispatchFence: predecessor.handle, + }, { + mergeAgent: squashMergeAgent(BRANCH), + reviewAgent: approveReviewAgent, + })).rejects.toBeInstanceOf(WorkspaceMergeDispatchSupersededError); + + expect(successorClaimed).toBe(true); + // FNXC:WorkspaceMergeDispatch 2026-08-15-10:09: the real store rejects before invoking the finalizer callback. + expect(finalizer).toHaveBeenCalledOnce(); + expect(terminalCallbackEntered).toBe(false); + expect((await store.getTask(taskId))!.column).toBe("in-review"); + expect((await store.getTask(taskId))!.mergeDetails).toBeUndefined(); + expect(fx.git("repo-a", "git rev-parse refs/heads/main")).not.toBe(tipBefore); + }); +}); + describeIfGit("landWorkspaceTask — per-repo land lease (Phase C U3, KTD4)", () => { let fx: WorkspaceFixture; afterEach(() => { @@ -142,6 +313,143 @@ describeIfGit("landWorkspaceTask — per-repo land lease (Phase C U3, KTD4)", () }); beforeEach(() => activeSessionRegistry.clear()); + it("rejects a superseded merge-dispatch pin even when the integration tip is unchanged", async () => { + fx = await createWorkspaceFixture(["repo-a"]); + const repo = fx.repoPath("repo-a"); + const remote = path.join(fx.rootDir, "origin.git"); + execSync(`git init --bare ${remote}`, { stdio: "pipe" }); + fx.git("repo-a", `git remote add origin ${remote}`); + fx.git("repo-a", "git push -u origin main"); + fx.git("repo-a", "git checkout -b fusion/fn-9059-fence"); + writeFileSync(path.join(repo, "fence.txt"), "fenced\n", "utf-8"); + fx.git("repo-a", "git add fence.txt && git commit -m 'fenced source'"); + + const tipSha = fx.git("repo-a", "git rev-parse origin/main"); + const sourceSha = fx.git("repo-a", "git rev-parse HEAD"); + const tree = fx.git("repo-a", "git mktree { + fx = await createWorkspaceFixture(["repo-a", "repo-b"]); + for (const repoRel of fx.repos) { + const remote = path.join(fx.rootDir, `${repoRel}.git`); + execSync(`git init --bare ${remote}`, { stdio: "pipe" }); + fx.git(repoRel, `git remote add origin ${remote}`); + fx.git(repoRel, "git push -u origin main"); + addRepoBranchWithEdit(fx, repoRel, "FN-9059", `${repoRel} dispatch fence\n`); + } + const task = makeTask("FN-9059", Object.fromEntries(fx.repos.map((repoRel) => [repoRel, { + worktreePath: fx.repoPath(repoRel), branch: BRANCH, + }]))); + const store = createStore(task); + let token = 0n; + Object.assign(store, { + acquireWorkspaceLease: vi.fn(async (input: any) => ({ + outcome: "acquired", + handle: { + leaseKey: input.leaseKey, kind: input.kind, owner: input.owner, + fenceToken: ++token, + }, + })), + recordWorkspaceLeaseFenceRef: vi.fn(async (input: any) => ({ + ...input.handle, fenceRefName: input.fenceRefName, fenceRefSha: input.fenceRefSha, + })), + releaseWorkspaceLease: vi.fn().mockResolvedValue(true), + recordWorkspaceLandIntent: vi.fn().mockResolvedValue({ outcome: "valid" }), + resolveWorkspaceLandIntent: vi.fn(async (input: any) => { + await input.persistLandedSha(); + return { outcome: "resolved" }; + }), + }); + const dispatchHandle = { + leaseKey: "merge-dispatch:FN-9059", kind: "merge-dispatch" as const, + owner: { taskId: "FN-9059", nodeId: "node-a", incarnationId: "inc-a" }, fenceToken: ++token, + }; + + /* + FNXC:WorkspaceMergeDispatch 2026-08-15-09:46: + A real workspace root is deliberately not a git checkout. Each sub-repo remote must receive + the dispatch ref before its own atomic push; publishing only at the root cannot fence either. + */ + const dispatchRef = "refs/fusion/merge-dispatch/FN-9059"; + const result = await landWorkspaceTask(store, task, fx.rootDir, { + workspaceDispatchFence: dispatchHandle, + }, { + mergeAgent: squashMergeAgent(BRANCH, () => { + /* + FNXC:WorkspaceMergeDispatch 2026-08-15-10:18: + The first repository's merge agent is reached only after dispatch fencing has covered every + target remote. This ordering catches a successor that otherwise protects repo-a while a + predecessor can still resume and push untouched repo-b. + */ + expect(fx.git("repo-b", `git ls-remote origin ${dispatchRef}`)).toMatch(/^[0-9a-f]{40,64}\s/); + }), + reviewAgent: approveReviewAgent, + }); + + expect(result.allLanded).toBe(true); + for (const repoRel of fx.repos) { + expect(fx.git(repoRel, `git ls-remote origin ${dispatchRef}`)).toMatch(/^[0-9a-f]{40,64}\s/); + } + }); + + it("does not finalize after a dispatch lease expires following a successful workspace push", async () => { + fx = await createWorkspaceFixture(["repo-a"]); + addRepoBranchWithEdit(fx, "repo-a", "FN-9059", "fenced terminal persist\n"); + const repoAbs = fx.repoPath("repo-a"); + const task = makeTask("FN-9059", { "repo-a": { worktreePath: repoAbs, branch: BRANCH } }); + const tipBefore = fx.git("repo-a", "git rev-parse main"); + const store = createStore(task); + const terminalWrite = vi.fn().mockRejectedValue(new Error("Workspace lease is no longer valid")); + Object.assign(store, { withValidWorkspaceLease: terminalWrite }); + + /* + FNXC:WorkspaceMergeDispatch 2026-08-15-09:37: + Simulate the renewal callback never firing: a successor has already reclaimed dispatch after + the ref advance. The guarded finalizer callback must never run, so stale work cannot mark the + task done even though the landed commit remains available to crash recovery. + */ + await expect(landWorkspaceTask(store, task, fx.rootDir, { + workspaceDispatchFence: { + leaseKey: "merge-dispatch:FN-9059", + kind: "merge-dispatch", + owner: { taskId: "FN-9059", nodeId: "node-b", incarnationId: "inc-b" }, + fenceToken: 1n, + }, + }, { mergeAgent: squashMergeAgent(BRANCH), reviewAgent: approveReviewAgent })).rejects.toBeInstanceOf(WorkspaceMergeDispatchSupersededError); + + expect(terminalWrite).toHaveBeenCalledOnce(); + expect(store.moveTaskCalls).toEqual([]); + expect(store.task.mergeDetails).toBeUndefined(); + expect(fx.git("repo-a", "git rev-parse main")).not.toBe(tipBefore); + }); + 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"); diff --git a/packages/engine/src/merge/merger-ai.ts b/packages/engine/src/merge/merger-ai.ts index c4fcdac188..1202239ca2 100644 --- a/packages/engine/src/merge/merger-ai.ts +++ b/packages/engine/src/merge/merger-ai.ts @@ -59,8 +59,12 @@ import { type MergeResult, type MergeTargetResolution, type Settings, + resolveEngineIncarnationId, + resolveEngineNodeId, type Task, - type TaskStore, resolveReviewColumns + type TaskStore, + type WorkspaceLeaseHandle, + resolveReviewColumns, } from "@fusion/core"; import { selectUserCommentsForAgentContext } from "../agents/agent-user-comments.js"; import { resolveTaskWorkingBranch } from "../worktree/worktree-names.js"; @@ -107,6 +111,7 @@ import cycle (merger-ai-worktree imports `MIN_TEMP_WORKTREE_REAP_AGE_MS` from se */ import { isRepoLanded, findProvenLandedCommit, FUSION_TASK_ID_TRAILER_KEY } from "./workspace-land-predicate.js"; import { persistWorkspaceRepoLandFailure } from "./workspace-land-failure.js"; +import { ensureTenancyFenceRef, mergeDispatchFenceRef, pushWithWorkspaceFence, WorkspaceFenceRefError, workspaceLandFenceRef } from "./workspace-fence-ref.js"; import { finalizeProvenAutoMergeTask } from "./auto-merge-finalization.js"; import { getCommitTaskOwnership, detectAlreadyLandedOnMain } from "./already-merged-detector.js"; import { resolveLegacyAiMergeRootPath } from "../worktree/worktree-paths.js"; @@ -642,16 +647,36 @@ export async function landSquash(input: { */ allowDirtyLocalCheckoutSync?: boolean; signal?: AbortSignal; + /** FNXC:Workspace 2026-08-15-08:36: Workspace lands advance the shared remote under a durable tenant fence before local sync. */ + workspaceFence?: { remote: string; fenceRefName: string; fenceRefSha: string }; + /** A task-scoped dispatch pin supplements the repo pin for a workspace merge body. */ + workspaceDispatchFence?: { fenceRefName: string; fenceRefSha: string }; }): Promise { - const { projectRootDir, mergeRoot, integrationBranch, tipSha, squashSha, taskId, audit, resolveConflicts, allowDirtyLocalCheckoutSync = false, signal } = input; + const { projectRootDir, mergeRoot, integrationBranch, tipSha, squashSha, taskId, audit, resolveConflicts, allowDirtyLocalCheckoutSync = false, signal, workspaceFence, workspaceDispatchFence } = input; const emit = (outcome: LocalSyncOutcome, extra: Record = {}) => audit.git({ type: "merge:ai-local-sync", target: integrationBranch, metadata: { taskId, outcome, squashSha, ...extra } }).catch(() => undefined); const currentBranch = await git(["rev-parse", "--abbrev-ref", "HEAD"], projectRootDir).catch(() => ""); + let sharedRefAdvanced = false; + const advanceSharedWorkspaceRef = async (): Promise => { + if (!workspaceFence || sharedRefAdvanced) return; + await pushWithWorkspaceFence({ + cwd: mergeRoot, + remote: workspaceFence.remote, + sourceSha: squashSha, + targetRef: `refs/heads/${integrationBranch}`, + expectedTargetSha: tipSha, + fenceRefName: workspaceFence.fenceRefName, + fenceRefSha: workspaceFence.fenceRefSha, + ...(workspaceDispatchFence ? { additionalFenceRefs: [workspaceDispatchFence] } : {}), + }); + sharedRefAdvanced = true; + }; // Case B — target not checked out here: bare CAS ref advance. if (currentBranch !== integrationBranch) { assertMergeGenerationOwned(signal, taskId); + await advanceSharedWorkspaceRef(); const adv = await advanceIntegrationBranchRef({ rootDir: mergeRoot, projectRootDir, integrationBranch, newSha: squashSha, expectedCurrentSha: tipSha, taskId, audit, @@ -704,6 +729,7 @@ export async function landSquash(input: { // stash hook failure). Don't risk `merge --ff-only` aborting/clobbering: // advance the ref atomically and leave the user's working tree as-is. assertMergeGenerationOwned(signal, taskId); + await advanceSharedWorkspaceRef(); const adv = await advanceIntegrationBranchRef({ rootDir: mergeRoot, projectRootDir, integrationBranch, newSha: squashSha, expectedCurrentSha: tipSha, taskId, audit, @@ -721,6 +747,7 @@ export async function landSquash(input: { // Fast-forward the checkout (and the branch ref) to the squash. assertMergeGenerationOwned(signal, taskId); + await advanceSharedWorkspaceRef(); if (!(await gitOk(["merge", "--ff-only", squashSha], projectRootDir))) { if (stashed) await gitOk(["stash", "pop"], projectRootDir); // restore the user's edits return { outcome: "concurrent", localSync: "skipped-other-branch" }; @@ -772,8 +799,9 @@ FNXC:Workspace 2026-06-21-23:40 (Phase C U1, KTD1): `runAiMerge`'s former inline clean-room closure: pre-merge prune (rooted at THIS repo) → mkdtemp clean room → `git worktree add --detach` → installWorktreeDependencies → mergeAndReview → landSquash → the concurrent-advance CAS retry loop → the -activeSessionRegistry register/unregister + cleanup-finally. It advances ONE local -integration ref (no remote push) and returns what landed. It deliberately does NOT +activeSessionRegistry register/unregister + cleanup-finally. Single-repo callers advance one local +integration ref; workspace callers pin a durable tenancy fence and atomically advance the shared +remote integration ref before reconciling their local ref. It deliberately does NOT move the task or write task-level mergeDetails — that task-global finalization (`finalizeMerged`/`finalizeTask`/`evaluateNoCommitsNoOpFinalize`) stays with the caller, so the same primitive is callable per sub-repo from `landWorkspaceTask` @@ -822,6 +850,10 @@ export interface LandRepoContext { resolvable in the engine process environment, and avoids unnecessary work. */ noCommitsExpected?: boolean; + /** FNXC:Workspace 2026-08-15-08:36: Present only for workspace sub-repos; it fences the durable intent and remote ref advance. */ + workspaceLand?: { handle: WorkspaceLeaseHandle; repoRelPath: string; remote: string }; + /** FNXC:WorkspaceMergeDispatch 2026-08-19-00:00: Task-level pin that fences every merge-body ref advance. */ + workspaceDispatchFence?: { fenceRefName: string; fenceRefSha: string }; store: TaskStore; } @@ -844,14 +876,13 @@ export type LandOneRepoResult = }; /** - * Land `branch` onto `integrationBranch`'s LOCAL ref in `repoRootDir` via a - * repo-scoped clean room, retrying on concurrent advance. No remote push. See - * the FNXC note above for the extraction contract. + * Land `branch` through a repo-scoped clean room, retrying on concurrent advance. + * Workspace contexts publish the shared integration ref under their durable fence; + * standalone contexts retain the local-only land contract. See the FNXC note above. */ -// FNXC:Workspace 2026-06-22-09:30 (Phase C review B12): `landOneRepo` takes its store access -// exclusively through the `ctx` callbacks (log/setStatus/audit) and pre-built agents — it never -// touches a TaskStore directly. The former leading `store` param was dead and misleading at the -// call sites (they looked like they forwarded a store the function ignored), so it was dropped. +// FNXC:Workspace 2026-08-15-08:36: `landOneRepo` receives the TaskStore through its context so +// workspace land intents are written immediately before the fenced shared-ref advance. The former +// leading store parameter stays absent; context keeps all task-bound dependencies explicit. export async function landOneRepo( repoRootDir: string, branch: string, @@ -1083,6 +1114,26 @@ export async function landOneRepo( if (!freshTask) throw new Error(`AI merge task ${taskId} disappeared before squash gates`); await enforceAiMergeSquashGates({ store, task: freshTask, taskId, mergeRoot, branch, tipSha, squashSha, settings, audit, log, repoRel: ctx.repoRel, repoKeys: ctx.repoKeys }); + // FNXC:Workspace 2026-08-15-08:36: Persist the recovery intent before the shared ref can + // move. A later reconciler can then settle an interrupted remote advance without re-squashing. + let workspaceFence: { remote: string; fenceRefName: string; fenceRefSha: string } | undefined; + if (ctx.workspaceLand) { + const { handle, repoRelPath, remote } = ctx.workspaceLand; + if (!handle.fenceRefName || !handle.fenceRefSha) { + throw new Error(`Workspace land lease ${handle.leaseKey} is missing its fence pin`); + } + await store.recordWorkspaceLandIntent({ + handle, + taskId, + repoRelPath, + remoteUrl: await git(["remote", "get-url", remote], repoRootDir), + integrationRef: `refs/heads/${integrationBranch}`, + intendedSha: squashSha, + expectedTip: tipSha, + }); + workspaceFence = { remote, fenceRefName: handle.fenceRefName, fenceRefSha: handle.fenceRefSha }; + } + // 4 + 5. Land the squash on the target branch and sync the user's // checkout (AI reconciles a conflicting restore). await setStatus("landing"); @@ -1091,6 +1142,8 @@ export async function landOneRepo( resolveConflicts: stashResolveAgent, allowDirtyLocalCheckoutSync: ctx.allowDirtyLocalCheckoutSync === true, signal, + workspaceFence, + workspaceDispatchFence: ctx.workspaceDispatchFence, }); if (landed.outcome === "concurrent") { if (advanceRetries < MAX_CONCURRENT_ADVANCE_RETRIES) { @@ -1930,11 +1983,19 @@ export class WorkspaceFinalizeBlockedError extends Error { } } +/** A fenced merge pushed successfully, but a successor owns terminal finalization. */ +export class WorkspaceMergeDispatchSupersededError extends Error { + constructor(public readonly taskId: string) { + super(`Workspace merge dispatch lease was superseded before finalization for ${taskId}`); + this.name = "WorkspaceMergeDispatchSupersededError"; + } +} + export async function landWorkspaceTask( store: TaskStore, task: Task, workspaceRootDir: string, - options: MergerOptions = {}, + options: MergerOptions & { workspaceDispatchFence?: WorkspaceLeaseHandle } = {}, deps: AgentDeps = {}, ): Promise { const taskId = task.id; @@ -1983,6 +2044,44 @@ export async function landWorkspaceTask( let allLanded = true; await setStatus("merging"); + try { + + let workspaceDispatchFence: { fenceRefName: string; fenceRefSha: string } | undefined; + const recordDispatchFence = (store as Partial).recordWorkspaceLeaseFenceRef; + if (options.workspaceDispatchFence && typeof recordDispatchFence === "function") { + /* + FNXC:WorkspaceMergeDispatch 2026-08-15-10:18: + A successor must publish its dispatch pin to EVERY workspace target remote before any land + sequence begins. Publishing only when each repository reaches its loop leaves a later remote + writable by a resumed predecessor; renewal callbacks are liveness-only, never correctness. + */ + try { + for (const repoRel of repoKeys) { + const ensuredDispatchFence = await ensureTenancyFenceRef({ + store, + handle: options.workspaceDispatchFence, + claimOutcome: "reentrant", + remote: "origin", + cwd: join(workspaceRootDir, repoRel), + fenceRefName: mergeDispatchFenceRef(taskId), + }); + options.workspaceDispatchFence = ensuredDispatchFence; + } + if (!options.workspaceDispatchFence.fenceRefName || !options.workspaceDispatchFence.fenceRefSha) { + throw new WorkspaceFenceRefError(`Workspace merge dispatch lease ${options.workspaceDispatchFence.leaseKey} has no fence pin`, "transport"); + } + workspaceDispatchFence = { + fenceRefName: options.workspaceDispatchFence.fenceRefName, + fenceRefSha: options.workspaceDispatchFence.fenceRefSha, + }; + } catch (error) { + if (error instanceof WorkspaceFenceRefError) { + throw new WorkspaceRepoLandBusyError(repoKeys[0] ?? "workspace", "workspace-merge-dispatch-fence", taskId); + } + throw error; + } + } + /* FNXC:Workspace 2026-06-22-04:10 (Phase C review A3 — status 'merging' must never leak): The busy-throw (WorkspaceRepoLandBusyError) and the persist-failure throw @@ -1994,7 +2093,6 @@ export async function landWorkspaceTask( first is safe — finalize overwrites it. This finally only clears the transient merge status; it does not move the task. */ - try { for (const repoRel of repoKeys) { throwIfAborted(options.signal, taskId); const entry = workspaceWorktrees[repoRel]; @@ -2083,11 +2181,58 @@ export async function landWorkspaceTask( if (landLeaseHolder && landLeaseHolder.taskId !== taskId) { throw new WorkspaceRepoLandBusyError(repoRel, landLeaseHolder.taskId, taskId); } - activeSessionRegistry.registerPath(repoRootDir, { - taskId, - kind: "workspace-repo-land", - ownerKey: WORKSPACE_REPO_LAND_OWNER_KEY, - }); + + /* + FNXC:Workspace 2026-08-15-08:36: + A process-local registry cannot serialize workspace landers on separate engine nodes. Claim the + repository's durable lease before registering locally, pin its fence ref once, and retain the + resulting handle through land-intent resolution. The remote push checks that same pin atomically + with the integration ref, so a superseded tenant cannot advance shared history after its TTL. + */ + let durableLandLease: WorkspaceLeaseHandle | undefined; + try { + const acquireWorkspaceLease = (store as Partial).acquireWorkspaceLease; + /* + FNXC:Workspace 2026-08-15-08:47: + The optional branch is solely for legacy structural in-memory stores used + by single-process tests. A real TaskStore has this API, and any error from + it remains a fail-closed land contention rather than a registry fallback. + */ + if (typeof acquireWorkspaceLease === "function") { + const claim = await acquireWorkspaceLease.call(store, { + leaseKey: `repo:${repoRel}`, + kind: "land", + owner: { taskId, nodeId: resolveEngineNodeId(), incarnationId: resolveEngineIncarnationId() }, + leaseMs: 5 * 60_000, + }); + if (claim.outcome === "conflict") { + throw new WorkspaceRepoLandBusyError(repoRel, claim.conflict.taskId, taskId); + } + durableLandLease = claim.handle; + durableLandLease = await ensureTenancyFenceRef({ + store, + handle: durableLandLease, + claimOutcome: claim.outcome, + remote: "origin", + cwd: repoRootDir, + fenceRefName: workspaceLandFenceRef(repoRel), + }); + } + } catch (error) { + if (durableLandLease) await store.releaseWorkspaceLease(durableLandLease).catch(() => undefined); + if (error instanceof WorkspaceRepoLandBusyError) throw error; + throw new WorkspaceRepoLandBusyError(repoRel, "durable-workspace-lease", taskId); + } + try { + activeSessionRegistry.registerPath(repoRootDir, { + taskId, + kind: "workspace-repo-land", + ownerKey: WORKSPACE_REPO_LAND_OWNER_KEY, + }); + } catch (error) { + if (durableLandLease) await store.releaseWorkspaceLease(durableLandLease).catch(() => undefined); + throw error; + } try { const landResult = await landOneRepo(repoRootDir, entry.branch, integrationBranch, { @@ -2102,6 +2247,8 @@ export async function landWorkspaceTask( noCommitsExpected: task.noCommitsExpected === true, repoRel, repoKeys, + ...(durableLandLease ? { workspaceLand: { handle: durableLandLease, repoRelPath: repoRel, remote: "origin" } } : {}), + ...(workspaceDispatchFence ? { workspaceDispatchFence } : {}), store, }); if (landResult.outcome === "landed") { @@ -2115,7 +2262,22 @@ export async function landWorkspaceTask( recorded as `landed` in the in-memory result first so the error payload is accurate. */ try { - await persistRepoLandedSha(store, taskId, repoRel, landResult.squashSha); + if (durableLandLease) { + const resolved = await store.resolveWorkspaceLandIntent({ + handle: durableLandLease, + taskId, + repoRelPath: repoRel, + expectedIntentFenceToken: durableLandLease.fenceToken, + resolution: "landed", + resolvedSha: landResult.squashSha, + persistLandedSha: () => persistRepoLandedSha(store, taskId, repoRel, landResult.squashSha), + }); + if (resolved.outcome !== "resolved") { + throw new Error(`Workspace land intent for ${repoRel} was not resolved (${resolved.outcome})`); + } + } else { + await persistRepoLandedSha(store, taskId, repoRel, landResult.squashSha); + } } catch (persistErr: unknown) { const pmsg = getErrorMessage(persistErr); await log(`AI merge (workspace): sub-repo ${repoRel} landed (${short(landResult.squashSha)}) but persisting landedSha FAILED: ${pmsg} — escalating to partial land so a retry can recover (ref already advanced; retry will skip via trailer ancestor-check)`); @@ -2143,6 +2305,11 @@ export async function landWorkspaceTask( // A WorkspacePartialLandError from the persist-failure window above must PROPAGATE // (the engine parks/retries). The outer try/finally below resets status first (A3). if (err instanceof WorkspacePartialLandError) throw err; + // A dispatch-fence publication failure is contention/transport at the resource boundary, + // not a sub-repo merge failure. This body never reached its fenced push. + if (err instanceof WorkspaceFenceRefError) { + throw new WorkspaceRepoLandBusyError(repoRel, "workspace-merge-dispatch-fence", taskId); + } const message = getErrorMessage(err); await log(`AI merge (workspace): sub-repo ${repoRel} land failed: ${message}`); await audit.git({ type: "merge:ai-no-branch", target: entry.branch, metadata: { taskId, kind: "workspace-repo-land-failed", repo: repoRel, error: message } }).catch(() => undefined); @@ -2165,6 +2332,11 @@ export async function landWorkspaceTask( if (held && held.taskId === taskId && held.ownerKey === WORKSPACE_REPO_LAND_OWNER_KEY) { activeSessionRegistry.unregisterPath(repoRootDir); } + if (durableLandLease) { + await store.releaseWorkspaceLease(durableLandLease).catch((releaseError: unknown) => { + aiMergeLog.warn(`${taskId}: durable workspace land lease release refused for ${repoRel}: ${getErrorMessage(releaseError)}`); + }); + } } } } finally { @@ -2224,7 +2396,38 @@ export async function landWorkspaceTask( await fence.write("lifecycle", () => store.moveTask(taskId, reboundColumn, { preserveProgress: true, moveSource: "engine" } as Parameters[2])); return { taskId, repos, allLanded, finalized: false, finalizeBlockedReason: reason }; } - const finalized = await finalizeWorkspaceTask(store, taskId, task, repos, fence); + /* + FNXC:WorkspaceMergeDispatch 2026-08-15-09:37: + Dispatch admission is not a licence to finalize. The sub-repo pushes may have completed while + this tenancy's renewal callback was stalled, so hold the owner+fence transaction lock over the + terminal merge-details write and move-to-done sequence. A reclaimed lease never invokes this + callback: its already-pushed commits remain recoverable through landedSha/intent evidence, but + this stale generation must not write the task outcome. Renewal only improves liveness. + */ + const finalize = () => finalizeWorkspaceTask(store, taskId, task, repos, fence); + const withValidDispatchLease = (store as Partial).withValidWorkspaceLease; + if (options.workspaceDispatchFence && typeof withValidDispatchLease === "function") { + try { + const finalized = await (withValidDispatchLease.call( + store, + options.workspaceDispatchFence, + async () => finalize(), + ) as Promise); + return { taskId, repos, allLanded, finalized }; + } catch (error) { + if (error instanceof Error && error.message === "Workspace lease is no longer valid") { + await audit.database({ + type: "workspace-lease:merge-completed-unrecorded" as Parameters[0]["type"], + target: taskId, + metadata: { taskId, outcome: "superseded-after-push", landedRepoCount: repos.filter((repo) => repo.status === "landed").length }, + }).catch(() => undefined); + await log("AI merge (workspace): dispatch lease was superseded after landing; left pushed refs intact for durable recovery"); + throw new WorkspaceMergeDispatchSupersededError(taskId); + } + throw error; + } + } + const finalized = await finalize(); return { taskId, repos, allLanded, finalized }; } return { taskId, repos, allLanded, finalized: false }; @@ -2263,12 +2466,34 @@ async function persistRepoLandedSha( ): Promise { // FNXC:Workspace 2026-08-15-06:45: a new landing is strictly after its revert boundary, // so clear that invalidation marker while retaining the fresh landedSha as normal proof. - await store.mergeWorkspaceWorktreeEntry( - taskId, - repoRel, - { landedSha, landFailure: undefined, revertBoundarySha: undefined }, - { requireExistingEntry: true }, - ); + const mergeWorkspaceEntry = (store as Partial).mergeWorkspaceWorktreeEntry; + if (typeof mergeWorkspaceEntry === "function") { + await mergeWorkspaceEntry.call( + store, + taskId, + repoRel, + { landedSha, landFailure: undefined, revertBoundarySha: undefined }, + { requireExistingEntry: true }, + ); + return; + } + + /* + FNXC:Workspace 2026-08-15-09:08: + Production stores use the advisory-locked per-repository merge above. Retain the + read/merge/write fallback only for structural single-process test stores that + intentionally predate that API; routing a real durable store through it would + reintroduce sibling-map clobbering. + */ + const task = await store.getTask(taskId); + const current = task.workspaceWorktrees?.[repoRel]; + if (!current) return; + await store.updateTask(taskId, { + workspaceWorktrees: { + ...task.workspaceWorktrees, + [repoRel]: { ...current, landedSha, landFailure: undefined, revertBoundarySha: undefined }, + }, + }); } /** diff --git a/packages/engine/src/merge/workspace-fence-ref.ts b/packages/engine/src/merge/workspace-fence-ref.ts new file mode 100644 index 0000000000..d870c0a5db --- /dev/null +++ b/packages/engine/src/merge/workspace-fence-ref.ts @@ -0,0 +1,171 @@ +import { execFile } from "node:child_process"; +import { createHash } from "node:crypto"; +import { promisify } from "node:util"; +import type { WorkspaceLeaseClaimOutcome, WorkspaceLeaseHandle } from "@fusion/core"; + +const execFileAsync = promisify(execFile); + +export class WorkspaceFenceRefError extends Error { + constructor( + message: string, + readonly kind: "cas-rejected" | "transport", + ) { + super(message); + this.name = "WorkspaceFenceRefError"; + } +} + +interface WorkspaceLeaseFenceStore { + recordWorkspaceLeaseFenceRef(input: { + handle: WorkspaceLeaseHandle; + fenceRefName: string; + fenceRefSha: string; + }): Promise; +} + +async function git(args: string[], cwd: string): Promise { + const { stdout } = await execFileAsync("git", args, { + cwd, + encoding: "utf8", + maxBuffer: 1024 * 1024, + }); + return stdout.trim(); +} + +async function remoteRefSha(remote: string, ref: string, cwd: string): Promise { + const output = await git(["ls-remote", remote, ref], cwd); + const sha = output.split(/\s+/)[0]; + return /^[0-9a-f]{40,64}$/i.test(sha ?? "") ? sha : undefined; +} + +/** Stable, collision-resistant slug without exposing a filesystem path as a ref hierarchy. */ +export function workspaceLandFenceRef(repoRelPath: string): string { + const slug = createHash("sha256").update(repoRelPath).digest("hex").slice(0, 24); + return `refs/fusion/workspace-lease/${slug}`; +} + +export function mergeDispatchFenceRef(taskId: string): string { + // Task ids are server-generated restricted identifiers; still normalize the ref component. + return `refs/fusion/merge-dispatch/${taskId.replaceAll(/[^A-Za-z0-9._-]/g, "_")}`; +} + +async function createFenceObject(cwd: string, handle: WorkspaceLeaseHandle): Promise { + /* + FNXC:WorkspaceMergeDispatch 2026-08-15-09:46: + A merge-dispatch tenancy can land into several independent sub-repo remotes. Its fence object + must therefore have the same object id in every clone. Pin both commit dates so this portable + private-ref commit is deterministic across those clones. + */ + const tree = await git(["hash-object", "-w", "-t", "tree", "/dev/null"], cwd); + const message = `workspace lease ${handle.leaseKey} ${handle.owner.taskId} ${handle.owner.nodeId} ${handle.owner.incarnationId} ${handle.fenceToken}`; + const { stdout } = await execFileAsync("git", [ + "-c", "user.name=Fusion", "-c", "user.email=fusion@localhost", "commit-tree", tree, "-m", message, + ], { + cwd, + encoding: "utf8", + env: { + ...process.env, + GIT_AUTHOR_DATE: "2000-01-01T00:00:00Z", + GIT_COMMITTER_DATE: "2000-01-01T00:00:00Z", + }, + }); + return stdout.trim(); +} + +/** + * FNXC:Workspace 2026-08-15-08:30: + * Fence refs are published once per tenancy, not once per claim call. A reentrant + * claim must reuse its recorded pin: rotating it would reject this tenancy's own + * prepared atomic push and orphan a pending land intent that names the old pin. + */ +export async function ensureTenancyFenceRef(input: { + store: WorkspaceLeaseFenceStore; + handle: WorkspaceLeaseHandle; + claimOutcome: WorkspaceLeaseClaimOutcome; + remote: string; + cwd: string; + fenceRefName: string; +}): Promise { + const { handle, claimOutcome } = input; + const publishRequired = claimOutcome === "acquired" + || claimOutcome === "reclaimed-expired" + || (claimOutcome === "reentrant" && !handle.fenceRefSha); + if (!publishRequired && (!handle.fenceRefName || !handle.fenceRefSha)) { + throw new WorkspaceFenceRefError(`Workspace lease ${handle.leaseKey} has no recorded fence pin`, "transport"); + } + + const observed = await remoteRefSha(input.remote, input.fenceRefName, input.cwd); + // Reentrant callers normally find their pin already present. A workspace has one dispatch + // tenancy but multiple remotes, so an absent/different remote must receive the same pin too. + if (handle.fenceRefSha && observed === handle.fenceRefSha) return handle; + const fenceSha = await createFenceObject(input.cwd, handle); + if (handle.fenceRefSha && fenceSha !== handle.fenceRefSha) { + throw new WorkspaceFenceRefError(`Workspace lease ${handle.leaseKey} fence object is not portable`, "transport"); + } + const expected = observed ?? ""; + try { + await git([ + "push", + `--force-with-lease=${input.fenceRefName}:${expected}`, + input.remote, + `${fenceSha}:${input.fenceRefName}`, + ], input.cwd); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + const kind = /stale info|rejected|lease|non-fast-forward/i.test(message) ? "cas-rejected" : "transport"; + throw new WorkspaceFenceRefError( + `Unable to publish required workspace fence ref ${input.fenceRefName}: ${message}`, + kind, + ); + } + + // A stored pin identifies the tenancy. Publishing it to another sub-repo remote does not + // rotate that durable pin or consume a new fence token. + if (handle.fenceRefSha) return handle; + try { + return await input.store.recordWorkspaceLeaseFenceRef({ + handle, + fenceRefName: input.fenceRefName, + fenceRefSha: fenceSha, + }); + } catch { + throw new WorkspaceFenceRefError(`Workspace lease ${handle.leaseKey} was superseded while recording fence pin`, "cas-rejected"); + } +} + +/** A single target+fence atomic CAS is the resource-level guard for shared git writes. */ +export async function pushWithWorkspaceFence(input: { + cwd: string; + remote: string; + sourceSha: string; + targetRef: string; + expectedTargetSha?: string; + fenceRefName: string; + fenceRefSha: string; + /** Extra resource pins required by a caller's enclosing tenancy. */ + additionalFenceRefs?: Array<{ fenceRefName: string; fenceRefSha: string }>; +}): Promise { + const targetExpected = input.expectedTargetSha ?? ""; + const fenceRefs = [ + { fenceRefName: input.fenceRefName, fenceRefSha: input.fenceRefSha }, + ...(input.additionalFenceRefs ?? []), + ]; + try { + await git([ + "push", + "--atomic", + `--force-with-lease=${input.targetRef}:${targetExpected}`, + ...fenceRefs.map((fence) => `--force-with-lease=${fence.fenceRefName}:${fence.fenceRefSha}`), + input.remote, + `${input.sourceSha}:${input.targetRef}`, + ...fenceRefs.map((fence) => `${fence.fenceRefSha}:${fence.fenceRefName}`), + ], input.cwd); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + const kind = /stale info|rejected|lease|non-fast-forward/i.test(message) ? "cas-rejected" : "transport"; + throw new WorkspaceFenceRefError( + `Workspace fenced atomic push failed for ${input.targetRef} using ${input.fenceRefName}: ${message}`, + kind, + ); + } +} diff --git a/packages/engine/src/project-engine.ts b/packages/engine/src/project-engine.ts index 181711d77e..035dbd74e2 100644 --- a/packages/engine/src/project-engine.ts +++ b/packages/engine/src/project-engine.ts @@ -48,6 +48,9 @@ import { clearMergeConfirmedTransientStatus, classifyGhError, createRecallCaptureWriter, + resolveEngineIncarnationId, + resolveEngineNodeId, + type WorkspaceLeaseHandle, } from "@fusion/core"; import { assemblePlannerOverseerRuntimeSnapshot } from "./overseer/planner-overseer-runtime-snapshot.js"; import { resolveIntegrationBranch } from "./merge/integration-branch.js"; @@ -83,6 +86,7 @@ import { runAiMerge, landWorkspaceTask, WorkspaceFinalizeBlockedError, + WorkspaceMergeDispatchSupersededError, WorkspacePartialLandError, WorkspaceRepoLandBusyError, } from "./merge/merger-ai.js"; @@ -100,6 +104,18 @@ import { unregisterProjectVerificationLimit, } from "./concurrency/verification-concurrency.js"; import { runtimeLog } from "./logger.js"; + +class WorkspaceMergeDispatchBusyError extends Error { + readonly retryable = true; + + constructor( + readonly holderTaskId: string, + readonly requestingTaskId: string, + ) { + super(`workspace merge dispatch is in progress for task ${holderTaskId}`); + this.name = "WorkspaceMergeDispatchBusyError"; + } +} import type { HeartbeatTriggerScheduler } from "./agent-heartbeat.js"; import { ResearchOrchestrator } from "./research/research-orchestrator.js"; import { ResearchRunDispatcher } from "./research/research-dispatcher.js"; @@ -537,6 +553,20 @@ export class ProjectEngine { private autostashSweepTimer: ReturnType | null = null; private mergeActiveReconcileTimer: ReturnType | null = null; + /* + FNXC:Workspace 2026-08-15-08:19: + Workspace dispatch now needs one funnel for process-local merge activity so durable + dispatch-lease cleanup can be attached without leaving a raw Set mutation behind. + Direct `mergeActive.add` or `mergeActive.delete` outside these helpers is a defect. + */ + private markMergeActive(taskId: string): void { + this.mergeActive.add(taskId); + } + + private clearMergeActive(taskId: string): void { + this.mergeActive.delete(taskId); + } + /* 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 @@ -565,6 +595,82 @@ export class ProjectEngine { this.workspaceBusyReenqueueTimers.add(timer); } + /** + * FNXC:WorkspaceMergeDispatch 2026-08-15-08:56: + * A workspace merge body is claimed only after dequeue, so another engine may + * retain an ordinary queued card but cannot begin a competing land sequence. + * Legacy structural stores intentionally lack this API; a present durable API + * that fails is fail-closed because its holder state is then unknown. + */ + private async withWorkspaceMergeDispatchLease( + store: TaskStore, + taskId: string, + cwd: string, + body: (dispatchFence?: WorkspaceLeaseHandle) => Promise, + ): Promise { + const acquire = (store as Partial).acquireWorkspaceLease; + if (typeof acquire !== "function") return body(); + + let claim: Awaited>>; + try { + claim = await acquire.call(store, { + leaseKey: `merge-dispatch:${taskId}`, + kind: "merge-dispatch", + owner: { + taskId, + nodeId: resolveEngineNodeId(), + incarnationId: resolveEngineIncarnationId(), + }, + leaseMs: 5 * 60_000, + }); + } catch { + throw new WorkspaceMergeDispatchBusyError("durable-workspace-lease", taskId); + } + if (claim.outcome === "conflict") { + throw new WorkspaceMergeDispatchBusyError(claim.conflict.taskId, taskId); + } + + let handle: WorkspaceLeaseHandle = claim.handle; + /* + FNXC:WorkspaceMergeDispatch 2026-08-15-09:46: + Workspace roots may be non-git directories and each sub-repo has its own remote. The body + publishes this tenancy's deterministic dispatch pin at each target remote immediately before + its fenced write; publishing here would pin only an unrelated root origin and is not a fence. + */ + let leaseLost = false; + const renew = (store as Partial).renewWorkspaceLease; + const timer = typeof renew === "function" + ? setInterval(() => { + void renew.call(store, handle, 5 * 60_000).then((renewed) => { + if (renewed) { + handle = renewed; + } else { + leaseLost = true; + this.abortActiveMerge(taskId, "workspace-merge-dispatch-lease-lost"); + } + }).catch(() => { + leaseLost = true; + this.abortActiveMerge(taskId, "workspace-merge-dispatch-lease-renewal-failed"); + }); + }, 60_000) + : undefined; + timer?.unref?.(); + + try { + const result = await body(handle); + if (leaseLost) { + throw new WorkspaceMergeDispatchBusyError("durable-workspace-lease", taskId); + } + return result; + } finally { + if (timer) clearInterval(timer); + const release = (store as Partial).releaseWorkspaceLease; + if (typeof release === "function") { + await release.call(store, handle).catch(() => undefined); + } + } + } + /** * Pending manual merge resolvers — keyed by taskId. * When `onMerge` is called, the task is enqueued like auto-merge but a @@ -732,11 +838,11 @@ export class ProjectEngine { if (this.activeMergeTaskId === taskId) { this.abortActiveMerge(taskId, "merge-enqueuer-reclaim"); } - this.mergeActive.delete(taskId); + this.clearMergeActive(taskId); return this.internalEnqueueMerge(taskId); }); this.runtime.setMergeActiveClearer?.((taskId) => { - this.mergeActive.delete(taskId); + this.clearMergeActive(taskId); }); // FNXC:Workspace 2026-06-22-16:40 (Phase D P1 TOCTOU): expose the in-memory merge pipeline // (mergeQueue + mergeActive) to the workspace self-healing reconcilers so they don't @@ -832,7 +938,7 @@ export class ProjectEngine { this.activeMergeSession.dispose(); this.activeMergeSession = null; } - this.mergeActive.delete(taskId); + this.clearMergeActive(taskId); this.activeMergeTaskId = null; this.activeMergeStartedAtMs = null; return true; @@ -931,10 +1037,26 @@ export class ProjectEngine { legitimately mid-dispatch. Because `mergeActive` lingers across the entire dequeue→rawMerge window, checking it in addition to `mergeQueue` closes that TOCTOU gap. */ - isMergePending(taskId: string): boolean { - return this.mergeActive.has(taskId) + async isMergePending(taskId: string): Promise { + if (this.mergeActive.has(taskId) || this.mergeQueue.includes(taskId) - || this.capacityDeferredMergeTaskIds.has(taskId); + || this.capacityDeferredMergeTaskIds.has(taskId)) return true; + + /* + FNXC:WorkspaceMergeDispatch 2026-08-15-12:00: + A remote engine's dispatch lease is invisible to queue-local state. An + unreadable durable probe is conservatively pending, never permission to + re-enqueue or reclaim workspace work that another node may be landing. + */ + const inspect = (this.runtime.getTaskStore() as Partial).inspectWorkspaceLeases; + if (typeof inspect !== "function") return false; + try { + const leases = await inspect.call(this.runtime.getTaskStore(), { taskId }); + const now = new Date().toISOString(); + return leases.some((lease) => lease.kind === "merge-dispatch" && lease.status === "held" && lease.expiresAt > now); + } catch { + return true; + } } /** @@ -1522,7 +1644,7 @@ export class ProjectEngine { const queuedTaskIds = [...this.mergeQueue]; this.mergeQueue.length = 0; for (const queuedTaskId of queuedTaskIds) { - this.mergeActive.delete(queuedTaskId); + this.clearMergeActive(queuedTaskId); } // Terminate active merge session @@ -2530,7 +2652,7 @@ export class ProjectEngine { this.activeMergeSession = null; } else if (!this.hasMergeResolvers(taskId)) { this.mergeQueue = this.mergeQueue.filter((queuedTaskId) => queuedTaskId !== taskId); - this.mergeActive.delete(taskId); + this.clearMergeActive(taskId); } resolver.reject(new Error(`Merge request for ${taskId} aborted`)); }; @@ -3073,12 +3195,12 @@ export class ProjectEngine { runtimeLog.warn( `internalEnqueueMerge(${taskId}): skipped — mergeActive entry is leaked (not queued, not active). Reconciling stale entry and retrying enqueue now.`, ); - this.mergeActive.delete(taskId); + this.clearMergeActive(taskId); } else { return false; } } - this.mergeActive.add(taskId); + this.markMergeActive(taskId); this.mergeQueue.push(taskId); void this.drainMergeQueue().catch((err: unknown) => { runtimeLog.error( @@ -3236,7 +3358,7 @@ export class ProjectEngine { for (const taskId of [...this.mergeActive]) { if (taskId === this.activeMergeTaskId) continue; if (this.mergeQueue.includes(taskId)) continue; - this.mergeActive.delete(taskId); + this.clearMergeActive(taskId); cleared++; } return cleared; @@ -4260,7 +4382,7 @@ export class ProjectEngine { const retryMs = settings.pollIntervalMs ?? 15_000; const stashedResolvers = this.takeMergeResolvers(taskId); const generation = this.startupGeneration; - this.mergeActive.delete(taskId); + this.clearMergeActive(taskId); this.capacityDeferredMergeTaskIds.add(taskId); const timer = setTimeout(() => { const deferred = this.capacityDeferredMerges.get(taskId); @@ -4415,6 +4537,7 @@ export class ProjectEngine { const mergeTask = await store.getTask(taskId).catch(() => null); const isWorkspaceMerge = !!mergeTask && isWorkspaceTask(mergeTask); if (isWorkspaceMerge) { + return this.withWorkspaceMergeDispatchLease(store, taskId, cwd, async (dispatchFence) => { // 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 @@ -4428,7 +4551,15 @@ export class ProjectEngine { store, mergeTask!, cwd, - { ...mergerOptions, allowDirtyLocalCheckoutSync: settings.merger?.allowDirtyLocalCheckoutSync === true }, + { + ...mergerOptions, + allowDirtyLocalCheckoutSync: settings.merger?.allowDirtyLocalCheckoutSync === true, + // FNXC:WorkspaceMergeDispatch 2026-08-19-00:00: + // Admission is not a licence to write: the per-task dispatch pin travels to + // every workspace ref advance so git rejects an expired predecessor even if + // renewal never ran and the target tip has not moved. + workspaceDispatchFence: dispatchFence, + }, ); if (!workspaceResult.allLanded) { // FNXC:Workspace 2026-06-22-05:10 (Phase C review B7): @@ -4473,6 +4604,7 @@ export class ProjectEngine { worktreeRemoved: false, branchDeleted: false, } as MergeResult; + }); } // FNXC:MergerUnification 2026-06-21-19:05: @@ -4595,7 +4727,12 @@ export class ProjectEngine { 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) { + const isWorkspaceBusyError = err instanceof WorkspaceRepoLandBusyError + || err instanceof WorkspaceMergeDispatchBusyError + // FNXC:WorkspaceMergeDispatch 2026-08-15-09:37: a stale generation that pushed + // before losing its lease must yield to durable recovery without failing the task. + || err instanceof WorkspaceMergeDispatchSupersededError; + if (isWorkspaceBusyError && hasManualResolver) { await store .logEntry(taskId, `Workspace sub-repo land busy (contention): ${errorMsg}`, "WorkspaceRepoLandBusy") .catch(() => undefined); @@ -4603,7 +4740,7 @@ export class ProjectEngine { continue; } - if (err instanceof WorkspaceRepoLandBusyError && !hasManualResolver) { + if (isWorkspaceBusyError && !hasManualResolver) { const busyCount = this.workspaceBusyReenqueues.get(taskId) ?? 0; await store .logEntry(taskId, `Workspace sub-repo land busy (contention): ${errorMsg}`, "WorkspaceRepoLandBusy") @@ -5315,7 +5452,7 @@ export class ProjectEngine { } this.clearActiveMergeClaim(taskId); this.mergeAbortController = null; - this.mergeActive.delete(taskId); + this.clearMergeActive(taskId); // If a manual merge was requested while this task was already in-flight, // the waiter(s) were set but not consumed above. Resolve them now. if (this.hasMergeResolvers(taskId)) { @@ -5475,7 +5612,7 @@ export class ProjectEngine { this.activeMergeTaskId !== task.id ) { runtimeLog.warn(`Auto-merge handoff (${task.id}): clearing stale mergeActive before enqueue`); - this.mergeActive.delete(task.id); + this.clearMergeActive(task.id); } this.internalEnqueueMerge(task.id); } catch (err: unknown) { @@ -5570,7 +5707,7 @@ export class ProjectEngine { const removedFromQueue = this.mergeQueue.length !== queueLengthBefore; if (removedFromQueue) { - this.mergeActive.delete(task.id); + this.clearMergeActive(task.id); runtimeLog.log(`Paused in-review task removed from merge queue: ${task.id}`); } @@ -5618,7 +5755,7 @@ export class ProjectEngine { if (removedFromQueue) { if (this.activeMergeTaskId !== task.id) { - this.mergeActive.delete(task.id); + this.clearMergeActive(task.id); } runtimeLog.log(`Soft-deleted task removed from merge queue: ${task.id}`); } diff --git a/packages/engine/src/runtimes/in-process-runtime.ts b/packages/engine/src/runtimes/in-process-runtime.ts index 14187995f8..c647bf92be 100644 --- a/packages/engine/src/runtimes/in-process-runtime.ts +++ b/packages/engine/src/runtimes/in-process-runtime.ts @@ -939,7 +939,7 @@ export class InProcessRuntime * ProjectEngine before `start()` via `setMergePendingProvider`. Used by the workspace * self-healing reconcilers to avoid re-dispatching / reclaiming a task mid-dequeue→rawMerge. */ - private mergePendingProvider?: (taskId: string) => boolean; + private mergePendingProvider?: (taskId: string) => boolean | Promise; /** Tracks whether startup recovery was intentionally deferred due to pause state. */ private startupRecoveryDeferred = false; /** Prevent duplicate unpause recovery dispatches from racing each other. */ @@ -2426,7 +2426,7 @@ export class InProcessRuntime this.activeMergeAborter = abortActiveMerge; } - setMergePendingProvider(isMergePending: (taskId: string) => boolean): void { + setMergePendingProvider(isMergePending: (taskId: string) => boolean | Promise): void { this.mergePendingProvider = isMergePending; } diff --git a/packages/engine/src/self-healing.ts b/packages/engine/src/self-healing.ts index 7586a6749b..99fa4e379f 100644 --- a/packages/engine/src/self-healing.ts +++ b/packages/engine/src/self-healing.ts @@ -39,8 +39,11 @@ import { type TaskMoveLanes, resolveColumnFlags, IN_REVIEW_STALL_DEADLOCK_LOG_PR pruneTaskLifecycleEvents, pruneGitHubCheckStatesAsync, resolveAgentActivityAttribution, + resolveEngineIncarnationId, + resolveEngineNodeId, } from "@fusion/core"; import { finalizePlanningSegment } from "@fusion/core"; +import type { WorkspaceLandIntent } from "@fusion/core"; import type { MeshLeaseManager } from "./project/mesh-lease-manager.js"; import { createLogger, schedulerLog } from "./logger.js"; import { @@ -436,7 +439,7 @@ export interface SelfHealingOptions { workspace-repo-land lease (the owner is mid-dispatch and is about to register that lease). Undefined = "not pending" (graceful when unwired); production always wires it. */ - isMergePending?: (taskId: string) => boolean; + isMergePending?: (taskId: string) => boolean | Promise; /** * Minimum blocker age before stale merge fan-out is cleared from downstream * blockedBy pointers. Must be >= staleMergingStatusMinAgeMs. @@ -1025,6 +1028,131 @@ export class SelfHealingManager extends SelfHealingGitEvidence { return { live, livePaths }; } + /** + * FNXC:Workspace 2026-08-15-08:55: + * A registry only observes this process. Before a self-healing backward move, + * also read durable leases for the task; an unreadable durable store is live so + * a node never re-enqueues work that another node may still be landing. + */ + private async isWorkspaceTaskLiveDurably(task: Task): Promise<{ live: boolean; livePaths: string[] }> { + const local = this.isWorkspaceTaskLive(task); + if (local.live) return local; + const inspectWorkspaceLeases = (this.store as Partial).inspectWorkspaceLeases; + /* + FNXC:Workspace 2026-08-15-08:47: + Legacy structural test stores have no durable backend and are intentionally + single-process. Once the durable probe exists, an error is unknown/liveness + true; only API absence may preserve the established local-fixture behavior. + */ + if (typeof inspectWorkspaceLeases !== "function") return local; + try { + const now = new Date().toISOString(); + const leases = await inspectWorkspaceLeases.call(this.store, { taskId: task.id }); + return { + ...local, + live: leases.some((lease) => lease.status === "held" && lease.expiresAt > now), + }; + } catch { + return { ...local, live: true }; + } + } + + /** + * FNXC:Workspace 2026-08-15-08:59: + * A pending land intent survives a process death after the fenced remote push but before its + * landed SHA is persisted. Fetch the declared remote ref and prove ancestry from that fresh + * remote graph; a local checkout or this node's registry is never recovery authority. + */ + protected async readWorkspaceLandIntentRemoteEvidence( + intent: WorkspaceLandIntent, + ): Promise<{ resolution: "landed"; resolvedSha: string } | { resolution: "not-landed" } | undefined> { + const repoRootDir = resolve(this.options.rootDir, intent.repoRelPath); + if (relative(this.options.rootDir, repoRootDir).startsWith("..") || isAbsolute(intent.repoRelPath)) return undefined; + try { + await execAsync( + `git fetch --no-tags --quiet ${shellQuote(intent.remoteUrl)} ${shellQuote(intent.integrationRef)}`, + { cwd: repoRootDir, timeout: 30_000 }, + ); + const { stdout } = await execAsync("git rev-parse --verify FETCH_HEAD", { cwd: repoRootDir, timeout: 30_000 }); + const remoteTip = stdout.trim(); + if (!remoteTip) return undefined; + try { + await execAsync( + `git merge-base --is-ancestor ${shellQuote(intent.intendedSha)} FETCH_HEAD`, + { cwd: repoRootDir, timeout: 30_000 }, + ); + return { resolution: "landed", resolvedSha: intent.intendedSha }; + } catch (error: unknown) { + if (typeof error === "object" && error !== null && "code" in error && (error as { code?: unknown }).code === 1) { + return { resolution: "not-landed" }; + } + return undefined; + } + } catch { + return undefined; + } + } + + /* + FNXC:Workspace 2026-08-15-08:59: + Durable land intents are deliberately reconciled independently of their original node. The + orphan resolver serializes against the repo lease and refuses a live replacement holder, while + remote reachability decides whether the interrupted push actually landed. Unknown evidence stays + pending rather than being converted into an unsafe retry. + */ + async reconcilePendingWorkspaceLandIntents(): Promise { + const listPending = (this.store as Partial).listPendingWorkspaceLandIntents; + const resolveOrphaned = (this.store as Partial).resolveOrphanedWorkspaceLandIntent; + const mergeEntry = (this.store as Partial).mergeWorkspaceWorktreeEntry; + if (typeof listPending !== "function" || typeof resolveOrphaned !== "function" || typeof mergeEntry !== "function") return 0; + + let reconciled = 0; + try { + const intents = await listPending.call(this.store, { limit: 100 }); + for (const intent of intents) { + const evidence = await this.readWorkspaceLandIntentRemoteEvidence(intent); + if (!evidence) continue; + const result = await resolveOrphaned.call(this.store, { + leaseKey: `repo:${intent.repoRelPath}`, + taskId: intent.taskId, + repoRelPath: intent.repoRelPath, + expectedIntentFenceToken: intent.fenceToken, + resolution: evidence.resolution, + ...(evidence.resolution === "landed" ? { + resolvedSha: evidence.resolvedSha, + persistLandedSha: async () => { + const task = await mergeEntry.call( + this.store, + intent.taskId, + intent.repoRelPath, + { landedSha: evidence.resolvedSha }, + { requireExistingEntry: true }, + ); + if (task.workspaceWorktrees?.[intent.repoRelPath]?.landedSha !== evidence.resolvedSha) { + throw new Error("Workspace land intent task entry is unavailable"); + } + }, + } : {}), + }); + if (result.outcome !== "resolved") continue; + reconciled++; + await createRunAuditor(this.store, { + runId: generateSyntheticRunId("self-healing-workspace-land-intent", intent.taskId), + agentId: "self-healing", + taskId: intent.taskId, + phase: "reconcile-workspace-land-intent", + }).database({ + type: "task:reconcile-workspace-land-intent", + target: intent.taskId, + metadata: { taskId: intent.taskId, repo: intent.repoRelPath, resolution: evidence.resolution }, + }).catch(() => undefined); + } + } catch (error: unknown) { + log.warn(`reconcilePendingWorkspaceLandIntents failed: ${error instanceof Error ? error.message : String(error)}`); + } + return reconciled; + } + /* FNXC:Workspace 2026-06-22-14:10 (Phase D review C — terminal-owner liveness for lease reclaim): A `workspace-repo-land` lease may only be reclaimed when its owning task ROW is demonstrably @@ -1592,6 +1720,24 @@ export class SelfHealingManager extends SelfHealingGitEvidence { { name: "stale-incomplete-review", fn: () => this.recoverStaleIncompleteReviewTasks().then(() => undefined) }, { name: "failed-pre-merge-steps", fn: () => this.recoverReviewTasksWithFailedPreMergeSteps().then(() => undefined) }, { name: "missing-worktree-review-failures", fn: () => this.recoverMissingWorktreeReviewFailures().then(() => undefined) }, + /* + FNXC:Workspace 2026-08-15-08:59: + Settle a pre-crash remote advance before stale merge recovery re-enqueues the task, so the + retry skips a proven landed repo instead of rebuilding and attempting its squash again. + */ + /* + FNXC:Workspace 2026-08-15-09:08: + Startup may reclaim only already-expired leases from an earlier incarnation. + A stable node id identifies a slot, not a process, so touching an unexpired + same-node row could release a live peer after a restart or misconfiguration. + */ + { name: "release-stale-workspace-leases", fn: async () => { + const releaseStale = (this.store as Partial).releaseStaleWorkspaceLeasesForNode; + if (typeof releaseStale === "function") { + await releaseStale.call(this.store, resolveEngineNodeId(), { currentIncarnationId: resolveEngineIncarnationId() }); + } + } }, + { name: "reconcile-pending-workspace-land-intents", fn: () => this.reconcilePendingWorkspaceLandIntents().then(() => undefined) }, { name: "interrupted-merging", fn: () => this.recoverInterruptedMergingTasks().then(() => undefined) }, { name: "wedged-active-merge", fn: () => this.recoverWedgedActiveMerge().then(() => undefined) }, { name: "transient-merge-failures", fn: () => this.recoverTransientMergeFailures().then(() => undefined) }, @@ -2706,6 +2852,11 @@ export class SelfHealingManager extends SelfHealingGitEvidence { { name: "reconcile-stranded-workflow-continuations", fn: () => this.reconcileStrandedWorkflowContinuations() }, { name: "recover-mergeable-review", fn: () => this.recoverMergeableReviewTasks() }, // FNXC:Workspace 2026-06-22-09:30 (Phase D U1) — workspace-mode reconcilers. + { name: "reconcile-expired-workspace-leases", fn: async () => { + const reconcileExpired = (this.store as Partial).reconcileExpiredWorkspaceLeases; + if (typeof reconcileExpired === "function") await reconcileExpired.call(this.store); + } }, + { name: "reconcile-pending-workspace-land-intents", fn: () => this.reconcilePendingWorkspaceLandIntents() }, { name: "reconcile-workspace-partial-lands", fn: () => this.reconcileWorkspacePartialLands() }, { name: "reclaim-phantom-workspace-land-leases", fn: () => this.reclaimPhantomWorkspaceLandLeases() }, { name: "reconcile-orphaned-workspace-worktrees", fn: () => this.reconcileOrphanedWorkspaceWorktrees() }, @@ -10062,7 +10213,7 @@ const movedTask = await this.store.moveTask(task.id, completeLane); continue; } // GUARD 3 — workspace-aware liveness: ANY active sub-repo path / process signal. - const liveness = this.isWorkspaceTaskLive(task); + const liveness = await this.isWorkspaceTaskLiveDurably(task); if (liveness.live) { await this.emitWorkspacePartialLandNoAction(task, "live-worktree", liveness.livePaths); continue; @@ -10083,7 +10234,7 @@ const movedTask = await this.store.moveTask(task.id, completeLane); `mergeActive` lingers across the whole window, so this guard closes the gap. Never moves the task backward; emits no-action and leaves the in-flight dispatch to finish. */ - if (this.options.isMergePending?.(task.id) === true) { + if (await this.options.isMergePending?.(task.id) === true) { await this.emitWorkspacePartialLandNoAction(task, "merge-pending", liveness.livePaths); continue; } @@ -10339,10 +10490,9 @@ const movedTask = await this.store.moveTask(task.id, completeLane); if (settings.globalPause || settings.enginePaused) return 0; const entries = activeSessionRegistry.entriesByKind("workspace-repo-land"); - if (entries.length === 0) return 0; - /* FNXC:Workspace 2026-08-15-04:11: resolve terminal lane vocabularies once per sweep, AFTER - the early return above, so a board with no leases pays nothing. See `isWorkspaceOwnerLive`. */ + /* FNXC:Workspace 2026-08-15-12:00: durable rows must be swept even on a + node with no local registry entries; local state cannot represent peers. */ const leaseOwnerCompleteColumns = await resolveProjectColumnsForRoles(this.store, ["complete"]); const leaseOwnerArchivedColumns = await resolveProjectColumnsForRoles(this.store, ["archived"]); @@ -10352,6 +10502,29 @@ const movedTask = await this.store.moveTask(task.id, completeLane); const now = Date.now(); let reclaimed = 0; + const inspectLeases = (this.store as Partial).inspectWorkspaceLeases; + const reclaimLease = (this.store as Partial).reclaimWorkspaceLease; + if (typeof inspectLeases === "function" && typeof reclaimLease === "function") { + try { + const leases = await inspectLeases.call(this.store); + for (const lease of leases) { + if ((lease.kind !== "land" && lease.kind !== "acquire") || lease.status !== "held") continue; + const ageMs = now - Date.parse(lease.acquiredAt); + if (!Number.isFinite(ageMs) || ageMs < staleFloorMs) continue; + if (await this.options.isMergePending?.(lease.owner.taskId) === true) continue; + const result = await reclaimLease.call(this.store, { + leaseKey: lease.leaseKey, + expectedOwner: lease.owner, + expectedFenceToken: lease.fenceToken, + requireTerminalOwner: true, + reason: "phantom-workspace-land-lease", + }); + if (result.outcome === "reclaimed") reclaimed++; + } + } catch (error: unknown) { + log.warn(`reclaimPhantomWorkspaceLandLeases durable sweep failed: ${error instanceof Error ? error.message : String(error)}`); + } + } for (const entry of entries) { try { const ageMs = now - entry.registeredAt; @@ -10370,7 +10543,7 @@ const movedTask = await this.store.moveTask(task.id, completeLane); Reclaiming now would yank the lease out from under a live land. Skip; the existing age-floor + terminal-owner guards still apply once the owner truly settles. */ - if (this.options.isMergePending?.(entry.taskId) === true) continue; + if (await this.options.isMergePending?.(entry.taskId) === true) continue; const owner = await this.store.getTask(entry.taskId).catch(() => null); const ownerColumn = owner?.column ?? "deleted"; @@ -10528,14 +10701,14 @@ const movedTask = await this.store.moveTask(task.id, completeLane); */ if (task.paused || task.userPaused || task.nextRecoveryAt || this.isWorkspaceTaskLive(task).live - || this.options.isMergePending?.(task.id) === true + || await this.options.isMergePending?.(task.id) === true || this.options.getActiveMergeTaskId?.() === task.id) continue; if (completeColumns.has(task.column) || task.column === "done") { candidates.push({ task, lane: "complete" }); continue; } const lane: Lane | null = task.deletedAt ? "soft-deleted" : task.status === "failed" ? "failed" : null; if (!lane) continue; const touched = Math.max(Date.parse(task.columnMovedAt ?? "") || 0, Date.parse(task.updatedAt ?? "") || 0, Date.parse(task.deletedAt ?? "") || 0); if (!touched || now - touched < TERMINAL_WORKSPACE_WORKTREE_TEARDOWN_MIN_IDLE_MS) continue; - if (this.isWorkspaceTaskLive(task).live || this.options.isMergePending?.(task.id) === true || this.options.getActiveMergeTaskId?.() === task.id) continue; + if (this.isWorkspaceTaskLive(task).live || await this.options.isMergePending?.(task.id) === true || this.options.getActiveMergeTaskId?.() === task.id) continue; candidates.push({ task, lane }); } if (!candidates.length) return 0; diff --git a/packages/engine/src/util/run-audit.ts b/packages/engine/src/util/run-audit.ts index 01d46d79b8..06296bbf8c 100644 --- a/packages/engine/src/util/run-audit.ts +++ b/packages/engine/src/util/run-audit.ts @@ -625,6 +625,8 @@ export type DatabaseMutationType = /* FNXC:Workspace 2026-06-22-09:30 (Phase D U1) — workspace-mode self-healing run-audit events. */ /** Metadata: { taskId, landedRepos: string[], unlandedRepos: string[], failedRepos: string[], action: "re-enqueue" | "park-failed", reason } */ | "task:reconcile-workspace-partial-land" + /** Metadata: { taskId, repo, resolution: "landed" | "not-landed" } */ + | "task:reconcile-workspace-land-intent" /** Metadata: { taskId, reason: "auto-merge-off" | "user-paused" | "live-worktree", livePaths: string[] } */ | "task:reconcile-workspace-partial-land-no-action" /** Metadata: { taskId, path, kind: "workspace-repo-land", registeredAt, ageMs, staleBindingAgeFloorMs, ownerColumn, ownerTerminalReason: "missing" | "complete" | "archived" | "deleted" | "failed" } */ diff --git a/packages/engine/src/worktree/worktree-acquisition.ts b/packages/engine/src/worktree/worktree-acquisition.ts index 1ed88e5665..1a916ea5bc 100644 --- a/packages/engine/src/worktree/worktree-acquisition.ts +++ b/packages/engine/src/worktree/worktree-acquisition.ts @@ -4,7 +4,7 @@ import { lstat, mkdir, readFile, readdir, realpath, rename, rm, stat, writeFile import { exec } from "node:child_process"; import { isAbsolute, join, relative, resolve } from "node:path"; import { promisify } from "node:util"; -import {acquireWorktreePathReservation, canonicalizeWorktreePath, type RunMutationContext, type Settings, type Task, type TaskStore, type SecretsStore} from "@fusion/core"; +import { acquireWorktreePathReservation, canonicalizeWorktreePath, resolveEngineIncarnationId, resolveEngineNodeId, type RunMutationContext, type Settings, type Task, type TaskStore, type SecretsStore, type WorkspaceLeaseHandle } from "@fusion/core"; import { generateWorktreeName, resolveTaskWorkingBranch, slugify } from "./worktree-names.js"; import { resolveTaskWorktreePathForBackend, resolveWorktreesDir, WORKTREE_RECOVERY_DIRNAME } from "./worktree-paths.js"; import { hydrateWorktreeDb } from "./worktree-db-hydrate.js"; @@ -1293,6 +1293,8 @@ export async function acquireWorkspaceRepoWorktree( assertInRootRepoRelPath(repoRelPath, sep, isAbsolute, normalize); const repoAbsPath = join(workspaceRootDir, repoRelPath); + let durableAcquireLease: WorkspaceLeaseHandle | undefined; + /* FNXC:WorkspaceWorktree 2026-06-22-00:00: A remembered per-repo worktree is only reusable if it still exists and is a registered git @@ -1325,6 +1327,39 @@ export async function acquireWorkspaceRepoWorktree( await store.logEntry(task.id, `Remembered workspace worktree for ${repoRelPath} is no longer usable; re-acquiring`, existing.worktreePath, runContext); } + /* + FNXC:Workspace 2026-08-15-08:50: + Registry-only acquire serialization is invisible to another engine sharing the + central database. Claim before the local lookup/register fast path; a durable + conflict (including this task on a different node incarnation) is retryable + busy rather than permission to overwrite another process's worktree state. + */ + try { + const acquireWorkspaceLease = (store as Partial).acquireWorkspaceLease; + /* + FNXC:Workspace 2026-08-15-08:47: + Production TaskStore instances always expose the durable API. Structural + in-memory test stores predate it and remain single-process fixtures, so they + retain the registry fast path; a present API that errors still fails closed. + */ + if (typeof acquireWorkspaceLease === "function") { + const claim = await acquireWorkspaceLease.call(store, { + leaseKey: `repo:${repoRelPath}`, + kind: "acquire", + owner: { taskId: task.id, nodeId: resolveEngineNodeId(), incarnationId: resolveEngineIncarnationId() }, + leaseMs: 5 * 60_000, + }); + if (claim.outcome === "conflict") { + throw new WorkspaceRepoAcquireBusyError(repoRelPath, claim.conflict.taskId, task.id); + } + durableAcquireLease = claim.handle; + } + } catch (error) { + if (error instanceof WorkspaceRepoAcquireBusyError) throw error; + // The database is authoritative for cross-node claims; unknown must never degrade to no holder. + throw new WorkspaceRepoAcquireBusyError(repoRelPath, "durable-workspace-lease", task.id); + } + /* FNXC:Workspace 2026-06-22-09:00: Run best-effort observability (task log + audit) for the NON-FATAL post-acquire @@ -1377,6 +1412,7 @@ export async function acquireWorkspaceRepoWorktree( } catch { // best-effort observability only — never mask the busy error } + if (durableAcquireLease) await store.releaseWorkspaceLease(durableAcquireLease).catch(() => undefined); throw err; } /* @@ -1580,6 +1616,11 @@ export async function acquireWorkspaceRepoWorktree( if (held && held.taskId === task.id && held.ownerKey === WORKSPACE_REPO_ACQUIRE_OWNER_KEY) { registry.unregisterPath(repoAbsPath); } + if (durableAcquireLease) { + await store.releaseWorkspaceLease(durableAcquireLease).catch((releaseError: unknown) => { + logger?.warn(`${task.id}: durable workspace acquire lease release refused: ${releaseError instanceof Error ? releaseError.message : String(releaseError)}`); + }); + } } }