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) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
2026-08-15 03:22:13 -07:00
parent 2f99a8fb59
commit 9fa8b386ee
25 changed files with 1901 additions and 63 deletions

View File

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

View File

@@ -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/<id>` 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.

View File

@@ -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/<repo-slug>` or per-merge `refs/fusion/merge-dispatch/<task-id>` 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 `<project>/.fusion/recovery/worktrees`, or to `<worktreesDir>/.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.
<!-- FNXC:MergerUnification 2026-08-09-12:04: Master-plan U0 made clean-room `runAiMerge` the sole production merge path. The legacy `aiMergeTask` auto-prerebase policy is retained but inert, so executor reused-base refresh must not describe it as live merger behavior. -->

View File

@@ -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/<repo-slug>` and a merge body publishes `refs/fusion/merge-dispatch/<task-id>`. 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.

View File

@@ -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);
});
});

View File

@@ -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<Partial<WorkspaceLeaseOwner>> = [
{ 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;

View File

@@ -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();
}

View File

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

View File

@@ -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 $$;

View File

@@ -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 };
});
}

View File

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

View File

@@ -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<TaskStoreEvents> {
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<T>(handle: WorkspaceLeaseHandle, fn: (tx: import("./postgres/data-layer.js").DbTransaction) => Promise<T>) { 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<typeof reclaimWorkspaceLeaseAsync>[1]) { return reclaimWorkspaceLeaseAsync(this, input); }
reconcileExpiredWorkspaceLeases() { return reconcileExpiredWorkspaceLeasesAsync(this); }
releaseStaleWorkspaceLeasesForNode(nodeId: string, options: { currentIncarnationId: string }) { return releaseStaleWorkspaceLeasesForNodeAsync(this, nodeId, options); }
recordWorkspaceLandIntent(input: Parameters<typeof recordWorkspaceLandIntentAsync>[1]) { return recordWorkspaceLandIntentAsync(this, input); }
listPendingWorkspaceLandIntents(filter: Parameters<typeof listPendingWorkspaceLandIntentsAsync>[1] = {}) { return listPendingWorkspaceLandIntentsAsync(this, filter); }
resolveWorkspaceLandIntent(input: Parameters<typeof resolveWorkspaceLandIntentAsync>[1]) { return resolveWorkspaceLandIntentAsync(this, input); }
resolveOrphanedWorkspaceLandIntent(input: Parameters<typeof resolveOrphanedWorkspaceLandIntentAsync>[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<TaskSymbolResolution> {
try {

View File

@@ -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<TaskStore, "getAsyncLayer">;
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<void> {
await tx.execute(sql`SELECT pg_advisory_xact_lock(hashtextextended(${`${projectId}:${leaseKey}`}, 0))`);
}
function ownerMatches(row: Pick<LeaseRow, "ownerTaskId" | "ownerNodeId" | "ownerIncarnationId" | "ownerRunId">, 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<AcquireWorkspaceLeaseResult> {
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<WorkspaceLeaseHandle | undefined> {
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<boolean> {
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<T>(store: TaskStore, current: WorkspaceLeaseHandle, fn: (tx: DbTransaction) => Promise<T>): Promise<T> {
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<WorkspaceLeaseHandle> {
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<WorkspaceLease[]> {
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<WorkspaceLandIntent[]> {
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<void> }, 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<void> }) {
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<void> }) {
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<string[]> {
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<string[]> {
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;
}

View File

@@ -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<WorkspaceLeaseClaimOutcome, "conflict">; 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<Task, "column" | "status"> | null | undefined): boolean {
return row != null && (row.column === "done" || row.status === "failed");
}

View File

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

View File

@@ -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<typeof SelfHealingManager>[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/<id>` 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<void> }) => {
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"]);

View File

@@ -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);
}
});
});

View File

@@ -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<Task>);
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<Task>);
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 </dev/null");
const landFenceSha = fx.git("repo-a", `git commit-tree ${tree} -m land-fence`);
const staleDispatchSha = fx.git("repo-a", `git commit-tree ${tree} -m stale-dispatch`);
const successorDispatchSha = fx.git("repo-a", `git commit-tree ${tree} -m successor-dispatch`);
const landRef = "refs/fusion/workspace-lease/test-repo";
const dispatchRef = "refs/fusion/merge-dispatch/FN-9059";
fx.git("repo-a", `git push origin ${landFenceSha}:${landRef} ${staleDispatchSha}:${dispatchRef}`);
fx.git("repo-a", `git push --force-with-lease=${dispatchRef}:${staleDispatchSha} origin ${successorDispatchSha}:${dispatchRef}`);
/*
FNXC:WorkspaceMergeDispatch 2026-08-19-00:00:
A dispatch lease renewal may never run while a merge body is suspended. The resource must
reject its atomic ref advance after a successor republishes the task fence, even when main
still equals the predecessor's observed tip.
*/
await expect(landSquash({
projectRootDir: repo,
mergeRoot: repo,
integrationBranch: "main",
tipSha,
squashSha: sourceSha,
taskId: "FN-9059",
audit: { git: vi.fn().mockResolvedValue(undefined) } as any,
workspaceFence: { remote: "origin", fenceRefName: landRef, fenceRefSha: landFenceSha },
workspaceDispatchFence: { fenceRefName: dispatchRef, fenceRefSha: staleDispatchSha },
})).rejects.toBeInstanceOf(WorkspaceFenceRefError);
expect(fx.git("repo-a", "git ls-remote origin refs/heads/main").split(/\s+/)[0]).toBe(tipSha);
});
it("publishes one dispatch tenancy pin to every sub-repo remote before fenced pushes", async () => {
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");

View File

@@ -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<LandResult> {
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<string, unknown> = {}) =>
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<void> => {
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<WorkspaceMergeResult> {
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<TaskStore>).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<TaskStore>).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<TaskStore["moveTask"]>[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<TaskStore>).withValidWorkspaceLease;
if (options.workspaceDispatchFence && typeof withValidDispatchLease === "function") {
try {
const finalized = await (withValidDispatchLease.call(
store,
options.workspaceDispatchFence,
async () => finalize(),
) as Promise<boolean>);
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<typeof audit.database>[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<void> {
// 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<TaskStore>).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 },
},
});
}
/**

View File

@@ -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<WorkspaceLeaseHandle>;
}
async function git(args: string[], cwd: string): Promise<string> {
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<string | undefined> {
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<string> {
/*
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<WorkspaceLeaseHandle> {
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<void> {
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,
);
}
}

View File

@@ -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<typeof setTimeout> | null = null;
private mergeActiveReconcileTimer: ReturnType<typeof setInterval> | 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<T>(
store: TaskStore,
taskId: string,
cwd: string,
body: (dispatchFence?: WorkspaceLeaseHandle) => Promise<T>,
): Promise<T> {
const acquire = (store as Partial<TaskStore>).acquireWorkspaceLease;
if (typeof acquire !== "function") return body();
let claim: Awaited<ReturnType<NonNullable<TaskStore["acquireWorkspaceLease"]>>>;
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<TaskStore>).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<TaskStore>).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<boolean> {
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<TaskStore>).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}`);
}

View File

@@ -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<boolean>;
/** 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<boolean>): void {
this.mergePendingProvider = isMergePending;
}

View File

@@ -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<boolean>;
/**
* 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<TaskStore>).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<number> {
const listPending = (this.store as Partial<TaskStore>).listPendingWorkspaceLandIntents;
const resolveOrphaned = (this.store as Partial<TaskStore>).resolveOrphanedWorkspaceLandIntent;
const mergeEntry = (this.store as Partial<TaskStore>).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<TaskStore>).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<TaskStore>).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<TaskStore>).inspectWorkspaceLeases;
const reclaimLease = (this.store as Partial<TaskStore>).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;

View File

@@ -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" } */

View File

@@ -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<TaskStore>).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)}`);
});
}
}
}