import * as NodeServices from "@effect/platform-node/NodeServices"; import { expect, it, vi } from "vite-plus/test"; import { it as effectIt } from "@effect/vitest"; import { CheckpointScopeId, MessageId, NodeId, ProviderSessionId, ProviderThreadId, ProviderTurnId, ProviderDriverKind, ProviderInstanceId, ProviderSetupError, RunAttemptId, RunId, ThreadId, ProjectId, type OrchestrationV2ThreadProjection, OrchestrationV2DomainEvent, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Path from "effect/Path"; import * as PlatformError from "effect/PlatformError"; import * as Schema from "effect/Schema"; import * as ServerConfig from "../config.ts"; import { withWorkspaceLease } from "../workspace/workspaceLease.ts"; import * as GitWorkflow from "../git/GitWorkflowService.ts"; import * as ProjectService from "../project/ProjectService.ts"; import * as ProviderAuthService from "../provider/ProviderAuthService.ts"; import * as ServerSettings from "../serverSettings.ts"; import * as ContextHandoffService from "./ContextHandoffService.ts"; import * as EventSink from "./EventSink.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as ProviderSessionManager from "./ProviderSessionManager.ts"; import * as ProviderTurnStart from "./ProviderTurnStartService.ts"; import * as RunExecutionService from "./RunExecutionService.ts"; import * as RuntimePolicy from "./RuntimePolicy.ts"; const isDomainEvent = Schema.is(OrchestrationV2DomainEvent); const managedWorktreesDir = "/tmp/provider-turn-start-worktrees"; // fs.realpath applies `..` and `.` before following symlinks. const normalize = (target: string) => { const parts: Array = []; for (const part of target.split("/")) { if (part === "..") parts.pop(); else if (part !== "" && part !== ".") parts.push(part); } return `/${parts.join("/")}`; }; const layerWorktreeLocation = (worktreesDir = managedWorktreesDir) => Layer.mergeAll( Path.layer, Layer.mock(ServerConfig.ServerConfig)({ worktreesDir } as ServerConfig.ServerConfig["Service"]), ServerSettings.layerTest(), ); /** Starts a run on a thread whose worktree folder is missing, with routing reads failing after. */ function startWithMissingWorktree( worktreePath: string, realPaths: Record = { [managedWorktreesDir]: managedWorktreesDir, [`${managedWorktreesDir}/project`]: `${managedWorktreesDir}/project`, }, ) { const threadId = ThreadId.make("thread_provider_turn_start_projection_failure"); const runId = RunId.make("run_provider_turn_start_projection_failure"); const attemptId = RunAttemptId.make("attempt_provider_turn_start_projection_failure"); const rootNodeId = NodeId.make("node_provider_turn_start_projection_failure"); const providerThreadId = ProviderThreadId.make( "provider_thread_provider_turn_start_projection_failure", ); const providerSessionId = ProviderSessionId.make( "provider_session_provider_turn_start_projection_failure", ); const messageId = MessageId.make("message_provider_turn_start_projection_failure"); const checkpointScopeId = CheckpointScopeId.make( "checkpoint_scope_provider_turn_start_projection_failure", ); const projection = { thread: { id: threadId, projectId: ProjectId.make("project_provider_turn_start_projection_failure"), branch: "feature/restore", worktreePath, }, runs: [ { id: runId, status: "starting", rootNodeId, activeAttemptId: attemptId, providerThreadId, userMessageId: messageId, ordinal: 2, }, ], nodes: [{ id: rootNodeId, checkpointScopeId }], attempts: [{ id: attemptId }], providerThreads: [{ id: providerThreadId, providerSessionId }], messages: [{ id: messageId, text: "Continue", attachments: [] }], checkpointScopes: [{ id: checkpointScopeId }], contextHandoffs: [], contextTransfers: [], turnItems: [], } as unknown as OrchestrationV2ThreadProjection; let projectionReadCount = 0; const writeIfRunCurrent = vi.fn(() => Effect.succeed({ committed: true, storedEvents: [] } as never), ); const startRootRun = vi.fn(() => Effect.void); const pruneWorktrees = vi.fn(() => Effect.void); const createWorktree = vi.fn(() => Effect.succeed({} as never)); const layer = ProviderTurnStart.layer.pipe( Layer.provide( Layer.mergeAll( Layer.mock(ContextHandoffService.ContextHandoffServiceV2)({}), Layer.mock(EventSink.EventSinkV2)({ writeIfRunCurrent }), IdAllocator.layer, Layer.succeed(FileSystem.FileSystem, { exists: () => Effect.succeed(false), // Paths that exist on the fake disk, mapped to where their symlinks lead. realPath: (target: string) => normalize(target) in realPaths ? Effect.succeed(realPaths[normalize(target)]) : Effect.fail( PlatformError.systemError({ _tag: "NotFound", module: "FileSystem", method: "realPath", pathOrDescriptor: target, }), ), } as never), layerWorktreeLocation(), Layer.mock(GitWorkflow.GitWorkflowService)({ pruneWorktrees, createWorktree }), Layer.mock(ProjectService.ProjectService)({ getById: () => Effect.succeed( Option.some({ workspaceRoot: "/tmp/provider-turn-start-project" } as never), ), }), Layer.mock(ProjectionStore.ProjectionStoreV2)({ getThreadRecords: () => { projectionReadCount += 1; return Effect.succeed(projection); }, getTurnStartContext: () => { projectionReadCount += 1; return Effect.succeed({ ...projection, hasConversation: projection.messages.some( (m) => m.role === "user" && (m.text.trim().toLowerCase() !== "/compact" || m.attachments.length > 0), ), }); }, getRuntimeRecoveryProjection: () => { projectionReadCount += 1; return Effect.fail( new ProjectionStore.ProjectionStoreReadError({ threadId, cause: "simulated inherited-background projection failure", }), ); }, }), Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({}), Layer.mock(ProviderAuthService.ProviderAuthService)({ tryHandlePromptCommand: () => Effect.succeed(false), }), Layer.mock(RunExecutionService.RunExecutionServiceV2)({ startRootRun }), Layer.mock(RuntimePolicy.RuntimePolicyV2)({}), ), ), ); const start = Effect.gen(function* () { return yield* (yield* ProviderTurnStart.ProviderTurnStartServiceV2) .start({ threadId, runId }) .pipe(Effect.flip); }).pipe(Effect.provide(layer), Effect.runPromise); return { start, pruneWorktrees, createWorktree, writeIfRunCurrent, startRootRun, projectionReads: () => projectionReadCount, }; } it.each([ { label: "//", worktreePath: `${managedWorktreesDir}/project/feature-restore`, recreated: true, }, { label: "the real path of a symlinked managed folder", worktreePath: "/real-worktrees/project/feature-restore", realPaths: { [managedWorktreesDir]: "/real-worktrees", "/real-worktrees": "/real-worktrees" }, recreated: true, }, { label: "a path outside the managed folder", worktreePath: "/tmp/provider-turn-start-project/.git/hooks", recreated: false, }, { label: "a symlink inside the managed folder that leads out", worktreePath: `${managedWorktreesDir}/project/hooks`, realPaths: { [managedWorktreesDir]: managedWorktreesDir, [`${managedWorktreesDir}/project`]: "/tmp/provider-turn-start-project/.git", }, recreated: false, }, { label: "a folder nested below a worktree", worktreePath: `${managedWorktreesDir}/project/feature-restore/link/hooks`, recreated: false, }, { label: "'..' out of the managed folder", worktreePath: `${managedWorktreesDir}/project/../../outside`, recreated: false, }, { // Left raw on purpose: git follows `link` before applying `..`. label: "'..' after a symlink inside the managed folder", worktreePath: `${managedWorktreesDir}/project/x/link/../../hooks`, realPaths: { [managedWorktreesDir]: managedWorktreesDir, [`${managedWorktreesDir}/project`]: `${managedWorktreesDir}/project`, [`${managedWorktreesDir}/project/x`]: `${managedWorktreesDir}/project/x`, [`${managedWorktreesDir}/project/x/link`]: "/tmp/provider-turn-start-project/.git/refs/heads", }, recreated: false, }, { label: "a relative path", worktreePath: "project/feature-restore", recreated: false }, ])("recreates a missing worktree only at the managed location: $label", async (input) => { const harness = startWithMissingWorktree(input.worktreePath, input.realPaths); await harness.start; expect(harness.createWorktree).toHaveBeenCalledTimes(input.recreated ? 1 : 0); }); function makeLocalCommandHarness(input: { readonly text: string; readonly worktreePath?: string; readonly workspaceRoot?: string; readonly worktreesDir?: string; readonly fileSystem?: FileSystem.FileSystem; readonly previousNativeSession?: boolean; readonly previousMessages?: ReadonlyArray; readonly logoutFailure?: string; readonly openFailure?: unknown; readonly ensureThreadFailure?: unknown; readonly historyReadFailureAfterFallback?: unknown; readonly interruptOpen?: boolean; readonly interruptRunBeforeOpenFailure?: boolean; readonly writeFailure?: unknown; readonly failReadsAfterRunning?: boolean; readonly previousTurn?: { readonly runStatus: "failed" | "completed"; readonly turnStatus: "running" | "completed" | "interrupted"; readonly differentThread?: boolean; readonly liveSession?: boolean; readonly interruptFailure?: boolean; readonly interruptCurrentRun?: boolean; }; }) { const now = DateTime.makeUnsafe("2026-09-04T12:00:00Z"); const threadId = ThreadId.make("thread-native-account-command"); const runId = RunId.make("run-native-account-command"); const rootNodeId = NodeId.make("root-native-account-command"); const attemptId = RunAttemptId.make("attempt-native-account-command"); const providerThreadId = ProviderThreadId.make("new-provider-thread"); const providerSessionId = ProviderSessionId.make("new-provider-session"); const oldProviderThreadId = ProviderThreadId.make("existing-native-provider-thread"); const oldInstanceId = ProviderInstanceId.make("antigravity-personal"); const newInstanceId = ProviderInstanceId.make("codex-personal"); const checkpointScopeId = CheckpointScopeId.make("scope-native-account-command"); const messageId = MessageId.make("message-native-account-command"); const run: OrchestrationV2ThreadProjection["runs"][number] = { id: runId, threadId, ordinal: 2, providerInstanceId: newInstanceId, modelSelection: { instanceId: newInstanceId, model: "gpt-5.4" }, providerThreadId, userMessageId: messageId, rootNodeId, activeAttemptId: attemptId, status: "starting", requestedAt: now, startedAt: null, completedAt: null, checkpointId: null, contextHandoffId: null, }; const providerThread: OrchestrationV2ThreadProjection["providerThreads"][number] = { id: providerThreadId, driver: ProviderDriverKind.make("codex"), providerInstanceId: newInstanceId, providerSessionId, appThreadId: threadId, ownerNodeId: null, nativeThreadRef: null, nativeConversationHeadRef: null, status: "not_loaded", firstRunOrdinal: 2, lastRunOrdinal: 2, handoffIds: [], forkedFrom: null, createdAt: now, updatedAt: now, }; const message: OrchestrationV2ThreadProjection["messages"][number] = { id: messageId, threadId, runId, nodeId: rootNodeId, role: "user", text: input.text, attachments: [], streaming: false, createdBy: "user", creationSource: "web", createdAt: now, updatedAt: now, }; let projection: OrchestrationV2ThreadProjection = { thread: { id: threadId, activeProviderThreadId: providerThreadId, projectId: ProjectId.make("project-native-account-command"), branch: input.worktreePath === undefined ? null : "feature/restore", worktreePath: input.worktreePath ?? null, } as OrchestrationV2ThreadProjection["thread"], runs: [ ...(input.previousNativeSession ? [ { ...run, id: RunId.make("previous-native-run"), ordinal: 1, status: "completed" as const, providerInstanceId: oldInstanceId, providerThreadId: oldProviderThreadId, }, ] : []), run, ], attempts: [ { id: attemptId, runId, rootNodeId, attemptOrdinal: 1, providerInstanceId: newInstanceId, providerThreadId, providerTurnId: null, reason: "initial", status: "pending", startedAt: null, completedAt: null, }, ], nodes: [ { id: rootNodeId, threadId, runId, parentNodeId: null, rootNodeId, kind: "root_turn", status: "pending", countsForRun: true, providerThreadId, providerTurnId: null, nativeItemRef: null, runtimeRequestId: null, checkpointScopeId, startedAt: null, completedAt: null, }, ], providerThreads: [ ...(input.previousNativeSession ? [ { ...providerThread, id: oldProviderThreadId, providerInstanceId: oldInstanceId, driver: ProviderDriverKind.make("antigravity"), lastRunOrdinal: 1, nativeThreadRef: { driver: ProviderDriverKind.make("antigravity"), nativeId: "existing-session", strength: "strong" as const, }, }, ] : []), providerThread, ], messages: [ ...(input.previousMessages ?? []).map((text, index) => ({ ...message, id: MessageId.make(`previous-message-${index}`), text, })), message, ], checkpointScopes: [ { id: checkpointScopeId, threadId, runId, nodeId: rootNodeId, parentScopeId: null, providerThreadId, kind: "root_run", ordinalWithinParent: 0, advancesAppRunCount: true, cwd: "/tmp/native-account-command", createdAt: now, }, ], providerSessions: [], providerTurns: [], contextHandoffs: [], contextTransfers: [], turnItems: [], visibleTurnItems: [], runtimeRequests: [], subagents: [], plans: [], checkpoints: [], updatedAt: now, }; if ("historyReadFailureAfterFallback" in input) { const nativeThreadRef = { driver: providerThread.driver, nativeId: "native-resume-thread", strength: "strong" as const, }; projection = { ...projection, providerThreads: projection.providerThreads.map((candidate) => candidate.id === providerThreadId ? { ...candidate, nativeThreadRef } : candidate, ), }; } const previousTurnId = ProviderTurnId.make("previous-provider-turn"); if (input.previousTurn !== undefined) { const previousRunId = RunId.make("previous-failed-run"); const previousAttemptId = RunAttemptId.make("previous-failed-attempt"); const previousProviderThreadId = input.previousTurn.differentThread ? oldProviderThreadId : providerThreadId; projection = { ...projection, runs: [ { ...run, id: previousRunId, ordinal: 1, status: input.previousTurn.runStatus, rootNodeId: NodeId.make("previous-root"), activeAttemptId: previousAttemptId, providerThreadId: previousProviderThreadId, completedAt: now, }, run, ], providerThreads: [ { ...providerThread, nativeThreadRef: { driver: providerThread.driver, nativeId: "existing-native-conversation", strength: "strong", }, }, ], providerTurns: [ { id: previousTurnId, providerThreadId: previousProviderThreadId, nodeId: NodeId.make("previous-root"), runAttemptId: previousAttemptId, ordinal: 1, nativeTurnRef: null, status: input.previousTurn.turnStatus, startedAt: now, completedAt: input.previousTurn.turnStatus === "running" ? null : now, }, ], }; } let providerBusy = input.previousTurn?.runStatus === "failed" && input.previousTurn.turnStatus === "running" && !input.previousTurn.differentThread && input.previousTurn.liveSession !== false; const interruptTurn = vi.fn((request: ProviderAdapter.ProviderAdapterV2InterruptInput) => Effect.gen(function* () { expect(request.providerTurnId).toBe(previousTurnId); expect(request.providerThread.id).toBe(providerThreadId); if (input.previousTurn?.interruptFailure) { return yield* new ProviderAdapter.ProviderAdapterInterruptError({ driver: providerThread.driver, providerThreadId, providerTurnId: request.providerTurnId, cause: "provider interrupt failed", }); } providerBusy = false; if (input.previousTurn?.interruptCurrentRun) interruptRun(); }), ); const events: Array = []; const interruptRun = () => { projection = { ...projection, runs: projection.runs.map((candidate) => candidate.id === runId ? { ...candidate, status: "interrupted", completedAt: now } : candidate, ), }; }; const ensureThread = vi.fn(() => Effect.sync(() => { if (input.interruptRunBeforeOpenFailure === true) interruptRun(); }).pipe( Effect.andThen( Effect.fail( new ProviderAdapter.ProviderAdapterEventStreamError({ driver: providerThread.driver, providerSessionId, cause: input.ensureThreadFailure, }), ), ), ), ); const resumeFallbackSession = { driver: providerThread.driver, resumeThread: () => Effect.fail( new ProviderAdapter.ProviderAdapterEventStreamError({ driver: providerThread.driver, providerSessionId, cause: "native thread is gone", }), ), ensureThread: () => Effect.succeed(providerThread), }; const open = vi.fn(() => input.interruptOpen === true ? Effect.interrupt : "historyReadFailureAfterFallback" in input ? Effect.succeed(resumeFallbackSession as never) : "ensureThreadFailure" in input ? Effect.succeed({ driver: providerThread.driver, ensureThread } as never) : "openFailure" in input ? Effect.sync(() => { if (input.interruptRunBeforeOpenFailure === true) interruptRun(); }).pipe( Effect.andThen( Effect.fail( new ProviderSessionManager.ProviderSessionOpenError({ instanceId: newInstanceId, providerSessionId, cause: input.openFailure, }), ), ), ) : input.failReadsAfterRunning === true || input.previousTurn !== undefined ? Effect.succeed({ driver: providerThread.driver, providerSession: { id: providerSessionId, driver: providerThread.driver, providerInstanceId: newInstanceId, status: "ready", cwd: "/tmp/native-account-command", model: null, capabilities: CodexProviderCapabilitiesV2, createdAt: now, updatedAt: now, lastError: null, }, ensureThread: () => Effect.succeed(providerThread), resumeThread: () => Effect.succeed(projection.providerThreads[0]), } as never) : Effect.die("A local command must not open a native session."), ); const startRootRun = vi.fn< (input: RunExecutionService.RunExecutionServiceV2StartRootRunInput) => Effect.Effect >(() => input.failReadsAfterRunning === true || input.previousTurn !== undefined ? Effect.suspend(() => providerBusy ? Effect.die("The previous provider turn is still running.") : Effect.void, ) : Effect.die("A local command must not start a native turn."), ); const failReadIfRunning = Effect.suspend(() => input.failReadsAfterRunning === true && projection.runs.find((candidate) => candidate.id === runId)?.status === "running" ? Effect.fail( new ProjectionStore.ProjectionStoreReadError({ threadId, cause: "database unavailable" }), ) : Effect.void, ); const tryHandlePromptCommand = vi.fn(() => input.logoutFailure === undefined ? Effect.succeed(true) : Effect.fail( new ProviderSetupError({ instanceId: oldInstanceId, operation: "logout", detail: input.logoutFailure, }), ), ); const writeIfRunCurrent = vi.fn(({ events: incoming, activeAttemptId, expectedStatus }) => "writeFailure" in input ? Effect.fail( new EventSink.EventSinkWriteError({ eventCount: incoming.length, cause: input.writeFailure, }), ) : Effect.sync(() => { const current = projection.runs.find((candidate) => candidate.id === runId); const committed = current !== undefined && current.activeAttemptId === activeAttemptId && current.status === expectedStatus; if (committed) { for (const event of incoming) { expect(isDomainEvent(event)).toBe(true); events.push(event); projection = ProjectionStore.applyToProjection(projection, event); } } return { committed, storedEvents: [] }; }), ); const resolveRuntimePolicy = vi.fn( ({ thread }: Parameters[0]) => Effect.succeed({ cwd: thread.worktreePath ?? "/tmp/native-account-command" } as never), ); const createWorktree = vi.fn(() => Effect.succeed({} as never)); const getThreadRecords = vi.fn(() => Effect.succeed(projection), ); const getTurnStartContext = vi.fn(() => Effect.succeed({ ...projection, hasConversation: projection.messages.some( (m) => m.role === "user" && (m.text.trim().toLowerCase() !== "/compact" || m.attachments.length > 0), ), }), ); const layer = ProviderTurnStart.layer.pipe( Layer.provide( Layer.mergeAll( Layer.mock(ContextHandoffService.ContextHandoffServiceV2)({ prepareProviderHandoff: () => Effect.die("history read must fail first"), }), Layer.mock(EventSink.EventSinkV2)({ writeIfRunCurrent }), IdAllocator.layer, layerWorktreeLocation(input.worktreesDir), input.fileSystem === undefined ? FileSystem.layerNoop({ exists: () => Effect.succeed(true) }) : Layer.succeed(FileSystem.FileSystem, input.fileSystem), Layer.mock(GitWorkflow.GitWorkflowService)({ pruneWorktrees: () => Effect.void, createWorktree, }), Layer.mock(ProjectService.ProjectService)({ getById: () => Effect.succeed(Option.some({ workspaceRoot: input.workspaceRoot ?? "/tmp/native-account-project" } as never)), }), Layer.mock(ProjectionStore.ProjectionStoreV2)({ getThreadRecords, getTurnStartContext, getRuntimeRecoveryProjection: () => Effect.as(failReadIfRunning, { ...projection, hasConversation: projection.messages.some( (m) => m.role === "user" && (m.text.trim().toLowerCase() !== "/compact" || m.attachments.length > 0), ), }), getTurnStartHistory: () => Effect.fail( new ProjectionStore.ProjectionStoreReadError({ threadId, cause: input.historyReadFailureAfterFallback, }), ), }), Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({ open, get: () => Effect.succeed( input.previousTurn?.liveSession === false ? Option.none() : Option.some({ driver: providerThread.driver, providerSessionId, interruptTurn, } as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime), ), }), Layer.mock(ProviderAuthService.ProviderAuthService)({ tryHandlePromptCommand }), Layer.mock(RunExecutionService.RunExecutionServiceV2)({ startRootRun }), Layer.mock(RuntimePolicy.RuntimePolicyV2)({ resolve: resolveRuntimePolicy, }), ), ), ); return { getThreadRecords, getTurnStartContext, setProjection: (next: OrchestrationV2ThreadProjection) => { projection = next; }, resolveRuntimePolicy, createWorktree, open, writeIfRunCurrent, startRootRun, interruptTurn, tryHandlePromptCommand, events, oldInstanceId, newInstanceId, attemptId, projection: () => projection, start: Effect.gen(function* () { yield* (yield* ProviderTurnStart.ProviderTurnStartServiceV2).start({ threadId, runId }); }).pipe(Effect.provide(layer)), startWithRetry: Effect.gen(function* () { yield* (yield* ProviderTurnStart.ProviderTurnStartServiceV2).start({ threadId, runId, willRetry: true, }); }).pipe(Effect.provide(layer)), }; } // --------------------------------------------------------------------------- // EVIDENCE DRIVER for PR #23671 (not part of the PR). // Service-layer driver on a real temp git worktree with an injected provider // pause; not a running server. // rollback side: real CheckpointRollbackServiceV2.execute (CheckpointRollbackService.layer), // ProviderSessionManagerV2.open parked on a Deferred, CheckpointServiceV2.restore runs a // real `git checkout -- file.txt` in a real temp git repo. // turn side: real ProviderTurnStartServiceV2.start via makeLocalCommandHarness (copied // verbatim from merge-base ProviderTurnStartService.test.ts, plus a workspaceRoot override). // Its FileSystem is the real NodeServices FileSystem with exists/realPath answered // synchronously from node:fs (real symlinks, real dirs) so the forked start either reaches // the provider or parks before the probe; see PROBE below. // Only public service APIs are called; no workspaceLease helpers are imported. // --------------------------------------------------------------------------- import * as NodeFs from "node:fs"; import * as NodeChildProcess from "node:child_process"; import { CheckpointId as DriverCheckpointId } from "@t3tools/contracts"; import * as DriverCheckpointService from "./CheckpointService.ts"; import * as DriverCheckpointRollbackService from "./CheckpointRollbackService.ts"; import * as DriverThreadCommandExecutor from "./ThreadCommandExecutor.ts"; import * as DriverProjectStore from "./ProjectStore.ts"; const PROBE_YIELDS = 64; const syncRealFileSystem = (real: FileSystem.FileSystem): FileSystem.FileSystem => ({ ...real, exists: (target) => Effect.sync(() => NodeFs.existsSync(target)), realPath: (target) => Effect.suspend(() => { try { return Effect.succeed(NodeFs.realpathSync(target)); } catch { return Effect.fail( PlatformError.systemError({ _tag: "NotFound", module: "FileSystem", method: "realPath", pathOrDescriptor: target, }), ); } }), }); const git = (cwd: string, ...args: Array) => NodeChildProcess.execFileSync("git", ["-C", cwd, ...args], { encoding: "utf8" }); type DriverCase = { readonly id: string; readonly label: string; readonly restoreFiles: boolean; readonly turn: (layout: { root: string; project: string; worktree: string; }) => { readonly worktreePath?: string; readonly workspaceRoot?: string }; }; const driverCases: ReadonlyArray = [ { id: "a", label: "symlink alias of the restoring worktree", restoreFiles: true, turn: ({ root }) => ({ worktreePath: `${root}/alias` }), }, { id: "b", label: "child dir inside the restoring worktree", restoreFiles: true, turn: ({ worktree }) => ({ worktreePath: `${worktree}/packages/app` }), }, { id: "c", label: "project-root thread (worktreePath null), workspaceRoot is a parent of the restoring worktree", restoreFiles: true, turn: ({ project }) => ({ workspaceRoot: project }), }, { id: "d", label: "CONTROL non-overlapping sibling dir ( vs -other)", restoreFiles: true, turn: ({ worktree }) => ({ worktreePath: `${worktree}-other` }), }, { id: "e", label: "CONTROL restoreFiles:false, turn in child dir of the rolling-back worktree", restoreFiles: false, turn: ({ worktree }) => ({ worktreePath: `${worktree}/packages/app` }), }, ]; effectIt.effect.each(driverCases)("EVIDENCE case ($id) $label", (driverCase) => Effect.gen(function* () { const realFs = yield* FileSystem.FileSystem; const path = yield* Path.Path; const t0 = performance.now(); const lines: Array = []; const log = (message: string) => lines.push( `[+${(performance.now() - t0).toFixed(3).padStart(8)}ms] #${String(lines.length + 1).padStart(2)} ${message}`, ); // Real temp layout: /project is the project; /worktrees/repo/feature is a // managed worktree (a real git repo); /alias -> worktree. const root = yield* realFs.realPath( yield* realFs.makeTempDirectoryScoped({ prefix: "t3-ev-23671-" }), ); const project = path.join(root, "project"); const worktreesDir = path.join(project, "worktrees"); const worktree = path.join(worktreesDir, "repo", "feature"); yield* realFs.makeDirectory(path.join(worktree, "packages", "app"), { recursive: true }); yield* realFs.makeDirectory(`${worktree}-other`, { recursive: true }); yield* realFs.symlink(worktree, path.join(root, "alias")); NodeFs.writeFileSync(path.join(worktree, "file.txt"), "checkpoint\n"); git(worktree, "init", "-q"); git(worktree, "-c", "user.email=ev@example.com", "-c", "user.name=ev", "add", "file.txt"); git( worktree, "-c", "user.email=ev@example.com", "-c", "user.name=ev", "commit", "-q", "-m", "checkpoint", ); NodeFs.writeFileSync(path.join(worktree, "file.txt"), "agent edit to roll back\n"); // ---- rollback side: real CheckpointRollbackServiceV2 ---- const threadId = ThreadId.make("ev-rollback-thread"); const providerThreadId = ProviderThreadId.make("ev-rollback-provider-thread"); const providerInstanceId = ProviderInstanceId.make("ev-rollback-provider"); const checkpointId = DriverCheckpointId.make("ev-rollback-checkpoint"); const scopeId = CheckpointScopeId.make("ev-rollback-scope"); const releaseProvider = Deferred.makeUnsafe(); const openEntered = Deferred.makeUnsafe(); const rollbackProjection = { thread: { id: threadId, projectId: ProjectId.make("ev-project"), worktreePath: worktree, branch: "feature/ev", activeProviderThreadId: providerThreadId, modelSelection: { instanceId: providerInstanceId, model: "test-model" }, }, providerThread: { id: providerThreadId, providerSessionId: ProviderSessionId.make("ev-rollback-session"), providerInstanceId, }, providerSession: undefined, checkpoint: { id: checkpointId, scopeId, status: "ready", appRunOrdinal: 0 }, scope: { id: scopeId, cwd: worktree }, runsToRollback: [], targetAttempt: undefined, rolledBackAttemptIds: [], staleCheckpoints: [], rootNodes: [], providerTurns: [], } as unknown as ProjectionStore.ProjectionCheckpointRollbackContext; const rollbackLayer = DriverCheckpointRollbackService.layer.pipe( Layer.provide( Layer.mergeAll( DriverThreadCommandExecutor.layer, NodeServices.layer, Layer.mock(ServerConfig.ServerConfig)({ worktreesDir } as never), ServerSettings.layerTest(), Layer.mock(DriverProjectStore.ProjectStoreV2)({ get: () => Effect.succeed(Option.some({ workspaceRoot: project } as never)), }), Layer.mock(GitWorkflow.GitWorkflowService)({}), Layer.mock(DriverCheckpointService.CheckpointServiceV2)({ restore: () => Effect.sync(() => { git(worktree, "checkout", "--", "file.txt"); const content = NodeFs.readFileSync(path.join(worktree, "file.txt"), "utf8"); log(`checkpoint files restored (git checkout; file.txt=${JSON.stringify(content)})`); }), }), Layer.mock(EventSink.EventSinkV2)({ write: () => Effect.succeed([]) }), IdAllocator.layer, Layer.mock(ProjectionStore.ProjectionStoreV2)({ getThread: () => Effect.succeed(rollbackProjection.thread), getCheckpointRollbackContext: () => Effect.succeed(rollbackProjection), getShellSnapshot: () => Effect.succeed({ schemaVersion: 1, snapshotSequence: 0, threads: [], archivedThreads: [], }), }), Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({ open: () => Deferred.succeed(openEntered, undefined).pipe( Effect.andThen(Deferred.await(releaseProvider)), Effect.as({} as never), ), }), Layer.mock(RuntimePolicy.RuntimePolicyV2)({ resolve: () => Effect.succeed({ cwd: worktree } as never), }), ), ), ); // ---- turn side: real ProviderTurnStartServiceV2 ---- const turn = driverCase.turn({ root, project, worktree }); const harness = makeLocalCommandHarness({ text: "Continue", ...turn, fileSystem: syncRealFileSystem(realFs), failReadsAfterRunning: true, }); const turnOpened = Deferred.makeUnsafe(); const open = harness.open.getMockImplementation()!; harness.open.mockImplementation(() => Effect.sync(() => log( `provider session opened for turn (worktree file.txt=${JSON.stringify(NodeFs.readFileSync(path.join(worktree, "file.txt"), "utf8"))})`, ), ).pipe( Effect.andThen(Deferred.succeed(turnOpened, undefined)), Effect.andThen(open()), ), ); lines.push(`case (${driverCase.id}) ${driverCase.label}`); lines.push( ` restore worktree=${worktree} restoreFiles=${driverCase.restoreFiles}; turn thread worktreePath=${turn.worktreePath ?? "null"}${turn.workspaceRoot === undefined ? "" : ` project.workspaceRoot=${turn.workspaceRoot}`}`, ); yield* Effect.gen(function* () { const service = yield* DriverCheckpointRollbackService.CheckpointRollbackServiceV2; const rollback = yield* service .execute({ threadId, providerThreadId, checkpointId, scopeId, restoreFiles: driverCase.restoreFiles, }) .pipe(Effect.forkChild({ startImmediately: true })); yield* Deferred.await(openEntered); log("rollback parked at provider open"); log("turn start requested"); const startup = yield* harness.start.pipe(Effect.forkChild({ startImmediately: true })); for (let i = 0; i < PROBE_YIELDS; i++) yield* Effect.yieldNow; const openedWhileParked = yield* Deferred.isDone(turnOpened); log( `PROBE after ${PROBE_YIELDS}x yieldNow: turn ${openedWhileParked ? "HAS opened a provider session" : "has NOT opened a provider session (waiting)"} while the rollback is parked`, ); yield* Deferred.succeed(releaseProvider, undefined); yield* Fiber.join(rollback); log("rollback released (execute returned)"); yield* Fiber.join(startup); log("turn start returned"); }).pipe(Effect.provide(rollbackLayer)); const restoredAt = lines.findIndex((line) => line.includes("checkpoint files restored")); const openedAt = lines.findIndex((line) => line.includes("provider session opened for turn")); const releasedAt = lines.findIndex((line) => line.includes("rollback released")); const probeAt = lines.findIndex((line) => line.includes("PROBE")); lines.push( `RESULT (${driverCase.id}): turn ${openedAt < probeAt ? "did NOT wait (opened while the rollback was parked)" : "WAITED for the rollback"}; ` + (restoredAt === -1 ? `no file restore (restoreFiles:false); provider session opened ${openedAt < releasedAt ? "BEFORE" : "AFTER"} the rollback released` : openedAt < restoredAt ? "provider session opened BEFORE checkpoint files restored" : "checkpoint files restored BEFORE provider session opened"), ); console.log(lines.join("\n")); expect(openedAt).toBeGreaterThan(-1); }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), );