feat(server): make rollback recovery durable

Refs #200

Author: rynfar

apps/server/src/git/GitWorkflowService.test.ts

@@ -2,12 +2,16 @@ import { assert, describe, expect, it, vi } from "@effect/vitest";
 import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
 
-import { VcsRepositoryDetectionError } from "@t3tools/contracts";
+import { ProjectId, ThreadId, VcsRepositoryDetectionError } from "@t3tools/contracts";
 
 import * as GitManager from "./GitManager.ts";
 import * as GitWorkflowService from "./GitWorkflowService.ts";
 import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
 import * as VcsDriverRegistry from "../vcs/VcsDriverRegistry.ts";
+import {
+  RollbackSagaRepository,
+  type RollbackSagaRecord,
+} from "../persistence/Services/RollbackSagas.ts";
 
 function makeLayer(input: {
   readonly detect: VcsDriverRegistry.VcsDriverRegistry["Service"]["detect"];
@@ -189,4 +193,77 @@ describe("GitWorkflowService", () => {
       ),
     );
   });
+
+  it.effect(
+    "fails closed before a Git mutation when the workspace rollback lease is active",
+    () => {
+      const threadId = ThreadId.make("thread-git-fence");
+      const projectId = ProjectId.make("project-git-fence");
+      const record = {
+        operationId: "operation-git-fence",
+        requestEventId: "event-git-fence",
+        threadId,
+        projectId,
+        workspaceKey: "workspace-git-fence",
+        phase: "workspace-apply-started",
+        terminal: false,
+        ownerId: null,
+        version: 1,
+        state: {
+          operationId: "operation-git-fence",
+          requestEventId: "event-git-fence",
+          threadId,
+          projectId,
+          workspaceKey: "workspace-git-fence",
+          workspaceCwd: "/repo",
+          sourceRevision: 2,
+          targetRevision: 1,
+          sourceCheckpointRef: "refs/source" as never,
+          sourceCheckpointOid: "a".repeat(40),
+          targetCheckpointRef: "refs/target" as never,
+          targetCheckpointOid: "b".repeat(40),
+          targetCheckpointDigest: "target-tree",
+          providerInstanceId: "fake" as never,
+          sessionIncarnationId: "session" as never,
+          phase: "workspace-apply-started" as const,
+          attempt: 0,
+          lastErrorCode: null,
+          compensation: "none" as const,
+          cleanup: "pending" as const,
+          sourceAnchor: null,
+          sourceAnchorDigest: null,
+          desiredAnchor: { leaf: "private" },
+          desiredAnchorDigest: "target",
+          preimage: { path: "private" },
+          workspaceReceiptDigest: null,
+          providerReceiptDigest: null,
+          projectionCommitSequence: null,
+          createdAt: "2026-08-31T00:00:00.000Z",
+          updatedAt: "2026-08-31T00:00:00.000Z",
+        },
+        createdAt: "2026-08-31T00:00:00.000Z",
+        updatedAt: "2026-08-31T00:00:00.000Z",
+      } satisfies RollbackSagaRecord;
+      const repository = Layer.succeed(RollbackSagaRepository, {
+        listNonterminal: () => Effect.succeed([record]),
+        listNonterminalForFence: () => Effect.succeed([record]),
+      } as never);
+      const testLayer = makeLayer({
+        detect: () => Effect.die("VCS detection must not run through a rollback fence"),
+      }).pipe(Layer.provideMerge(repository));
+
+      return Effect.gen(function* () {
+        const workflow = yield* GitWorkflowService.GitWorkflowService;
+        const result = yield* workflow.pullCurrentBranch("/repo").pipe(Effect.result);
+        assert.equal(result._tag, "Failure");
+        if (result._tag === "Failure") {
+          expect(result.failure).toMatchObject({
+            _tag: "GitCommandError",
+            command: "rollback-fence",
+            cwd: "/repo",
+          });
+        }
+      }).pipe(Effect.provide(testLayer));
+    },
+  );
 });

apps/server/src/git/GitWorkflowService.ts

@@ -1,6 +1,9 @@
+// @effect-diagnostics nodeBuiltinImport:off
 import * as Context from "effect/Context";
 import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
+import * as Option from "effect/Option";
+import * as NodeFS from "node:fs";
 
 import {
   GitManagerError,
@@ -31,6 +34,15 @@ import {
 import * as GitManager from "./GitManager.ts";
 import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
 import * as VcsDriverRegistry from "../vcs/VcsDriverRegistry.ts";
+import { RollbackSagaRepository } from "../persistence/Services/RollbackSagas.ts";
+
+function canonicalWorkspacePath(cwd: string): string {
+  try {
+    return NodeFS.realpathSync(cwd);
+  } catch {
+    return cwd;
+  }
+}
 
 export class GitWorkflowService extends Context.Service<
   GitWorkflowService,
@@ -141,6 +153,44 @@ export const make = Effect.gen(function* () {
   const registry = yield* VcsDriverRegistry.VcsDriverRegistry;
   const git = yield* GitVcsDriver.GitVcsDriver;
   const gitManager = yield* GitManager.GitManager;
+  const rollbackRepository = yield* Effect.serviceOption(RollbackSagaRepository);
+
+  const workspaceIsRollbackFenced = Effect.fn("GitWorkflowService.workspaceIsRollbackFenced")(
+    function* (cwd: string) {
+      if (Option.isNone(rollbackRepository)) return false;
+      const canonical = canonicalWorkspacePath(cwd);
+      const active = yield* rollbackRepository.value
+        .listNonterminalForFence()
+        .pipe(Effect.orElseSucceed(() => null));
+      if (active === null) return true;
+      return active.some((record) => record.state.workspaceCwd === canonical);
+    },
+  );
+  const ensureMutationCommand = Effect.fn("GitWorkflowService.ensureMutationCommand")(function* (
+    operation: string,
+    cwd: string,
+  ) {
+    if (yield* workspaceIsRollbackFenced(cwd)) {
+      return yield* new GitCommandError({
+        operation,
+        command: "rollback-fence",
+        cwd,
+        detail: "The workspace is fenced by an active rollback operation.",
+      });
+    }
+  });
+  const ensureMutationWorkflow = Effect.fn("GitWorkflowService.ensureMutationWorkflow")(function* (
+    operation: string,
+    cwd: string,
+  ) {
+    if (yield* workspaceIsRollbackFenced(cwd)) {
+      return yield* new GitManagerError({
+        operation,
+        cwd,
+        detail: "The workspace is fenced by an active rollback operation.",
+      });
+    }
+  });
 
   const ensureGit = Effect.fn("GitWorkflowService.ensureGit")(function* (
     operation: string,
@@ -281,29 +331,33 @@ export const make = Effect.gen(function* () {
     invalidateRemoteStatus: gitManager.invalidateRemoteStatus,
     invalidateStatus: gitManager.invalidateStatus,
     pullCurrentBranch: (cwd) =>
-      ensureGitCommand("GitWorkflowService.pullCurrentBranch", cwd).pipe(
+      ensureMutationCommand("GitWorkflowService.pullCurrentBranch", cwd).pipe(
+        Effect.andThen(ensureGitCommand("GitWorkflowService.pullCurrentBranch", cwd)),
         Effect.andThen(git.pullCurrentBranch(cwd)),
       ),
     runStackedAction: (input, options) =>
-      ensureGit("GitWorkflowService.runStackedAction", input.cwd).pipe(
+      ensureMutationWorkflow("GitWorkflowService.runStackedAction", input.cwd).pipe(
+        Effect.andThen(ensureGit("GitWorkflowService.runStackedAction", input.cwd)),
         Effect.andThen(gitManager.runStackedAction(input, options)),
       ),
     resolvePullRequest: routeGitManager(
       "GitWorkflowService.resolvePullRequest",
       gitManager.resolvePullRequest,
     ),
-    preparePullRequestThread: routeGitManager(
-      "GitWorkflowService.preparePullRequestThread",
-      gitManager.preparePullRequestThread,
-    ),
+    preparePullRequestThread: (input) =>
+      ensureMutationWorkflow("GitWorkflowService.preparePullRequestThread", input.cwd).pipe(
+        Effect.andThen(ensureGit("GitWorkflowService.preparePullRequestThread", input.cwd)),
+        Effect.andThen(gitManager.preparePullRequestThread(input)),
+      ),
     listRefs: (input) =>
       detectGitRepositoryForCommand("GitWorkflowService.listRefs", input.cwd).pipe(
         Effect.flatMap((isGitRepository) =>
           isGitRepository ? git.listRefs(input) : Effect.succeed(nonRepositoryListRefs()),
         ),
       ),
     createWorktree: (input) =>
-      ensureGitCommand("GitWorkflowService.createWorktree", input.cwd).pipe(
+      ensureMutationCommand("GitWorkflowService.createWorktree", input.cwd).pipe(
+        Effect.andThen(ensureGitCommand("GitWorkflowService.createWorktree", input.cwd)),
         Effect.andThen(git.createWorktree(input)),
       ),
     fetchRemote: (input) =>
@@ -319,23 +373,27 @@ export const make = Effect.gen(function* () {
         Effect.andThen(git.resolveRemoteTrackingCommit(input)),
       ),
     removeWorktree: (input) =>
-      ensureGitCommand("GitWorkflowService.removeWorktree", input.cwd).pipe(
+      ensureMutationCommand("GitWorkflowService.removeWorktree", input.cwd).pipe(
+        Effect.andThen(ensureGitCommand("GitWorkflowService.removeWorktree", input.cwd)),
         Effect.andThen(git.removeWorktree(input)),
       ),
     pruneWorktrees: (input) =>
       ensureGitCommand("GitWorkflowService.pruneWorktrees", input.cwd).pipe(
         Effect.andThen(git.pruneWorktrees(input)),
       ),
     createRef: (input) =>
-      ensureGitCommand("GitWorkflowService.createRef", input.cwd).pipe(
+      ensureMutationCommand("GitWorkflowService.createRef", input.cwd).pipe(
+        Effect.andThen(ensureGitCommand("GitWorkflowService.createRef", input.cwd)),
         Effect.andThen(git.createRef(input)),
       ),
     switchRef: (input) =>
-      ensureGitCommand("GitWorkflowService.switchRef", input.cwd).pipe(
+      ensureMutationCommand("GitWorkflowService.switchRef", input.cwd).pipe(
+        Effect.andThen(ensureGitCommand("GitWorkflowService.switchRef", input.cwd)),
         Effect.andThen(Effect.scoped(git.switchRef(input))),
       ),
     renameBranch: (input) =>
-      ensureGit("GitWorkflowService.renameBranch", input.cwd).pipe(
+      ensureMutationWorkflow("GitWorkflowService.renameBranch", input.cwd).pipe(
+        Effect.andThen(ensureGit("GitWorkflowService.renameBranch", input.cwd)),
         Effect.andThen(git.renameBranch(input)),
       ),
   });