diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts index 858f0996cdb69..5dd8cb00ba360 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -295,6 +295,7 @@ function makeLocalCommandHarness(input: { readonly resolveKnowledge?: ProjectKnowledge.ProjectKnowledge["Service"]["resolveForTurn"]; readonly readSentDeliveries?: ProjectKnowledge.ProjectKnowledge["Service"]["readSentDeliveries"]; readonly worktreePath?: string; + readonly branch?: string; readonly worktreesDir?: string; readonly workspaceRoot?: string; readonly gitWorkflow?: GitWorkflow.GitWorkflowService["Service"]; @@ -396,7 +397,7 @@ function makeLocalCommandHarness(input: { id: threadId, activeProviderThreadId: providerThreadId, projectId: ProjectId.make("project-native-account-command"), - branch: input.worktreePath === undefined ? null : "feature/restore", + branch: input.worktreePath === undefined ? null : (input.branch ?? "feature/restore"), worktreePath: input.worktreePath ?? null, lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, } as OrchestrationV2ThreadProjection["thread"], @@ -1317,6 +1318,53 @@ effectIt.effect( }).pipe(Effect.scoped, Effect.provide(layerRealGit)), ); +effectIt.effect("recreates a rebound thread on the branch its agent switched to", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const gitWorkflow = yield* GitWorkflow.GitWorkflowService; + const driver = yield* GitVcsDriver.GitVcsDriver; + const root = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-recreate-rebound-" }); + const workspaceRoot = path.join(root, "project"); + const worktreesDir = path.join(root, "worktrees"); + const worktreePath = path.join(worktreesDir, "project", "restore"); + yield* fileSystem.makeDirectory(workspaceRoot); + const git = (cwd: string, args: ReadonlyArray) => + driver + .execute({ operation: "test.worktreeRebound", cwd, args }) + .pipe(Effect.map((result) => result.stdout.trim())); + yield* git(workspaceRoot, ["init", "-b", "main"]); + yield* git(workspaceRoot, ["config", "user.name", "Test"]); + yield* git(workspaceRoot, ["config", "user.email", "test@example.com"]); + yield* git(workspaceRoot, ["commit", "--allow-empty", "-m", "base"]); + yield* git(workspaceRoot, ["worktree", "add", "-b", "feature/restore", worktreePath]); + yield* git(worktreePath, ["switch", "-c", "agent/work"]); + yield* git(worktreePath, ["commit", "--allow-empty", "-m", "agent work"]); + const agentTip = yield* git(worktreePath, ["rev-parse", "HEAD"]); + yield* fileSystem.remove(worktreePath, { recursive: true }); + const harness = makeLocalCommandHarness({ + text: "Continue", + branch: "agent/work", + workspaceRoot, + worktreePath, + worktreesDir, + fileSystem, + gitWorkflow, + failReadsAfterRunning: true, + }); + const open = harness.open.getMockImplementation()!; + harness.open.mockImplementation(() => + Effect.gen(function* () { + expect(yield* git(worktreePath, ["branch", "--show-current"])).toBe("agent/work"); + expect(yield* git(worktreePath, ["rev-parse", "HEAD"])).toBe(agentTip); + return yield* open(); + }).pipe(Effect.orDie), + ); + yield* harness.start; + expect(harness.open).toHaveBeenCalledOnce(); + }).pipe(Effect.scoped, Effect.provide(layerRealGit)), +); + effectIt.effect( "preserves recreation interrupts and retries, then fails the final attempt before provider start", () => diff --git a/apps/server/src/orchestration-v2/RunFinalizationQueue.test.ts b/apps/server/src/orchestration-v2/RunFinalizationQueue.test.ts index 004a7ea508c6b..b94d6f69274e4 100644 --- a/apps/server/src/orchestration-v2/RunFinalizationQueue.test.ts +++ b/apps/server/src/orchestration-v2/RunFinalizationQueue.test.ts @@ -23,6 +23,7 @@ import * as RunFinalization from "./RunFinalizationService.ts"; import * as RuntimeRequests from "./RuntimeRequestService.ts"; import * as ThreadManagement from "./ThreadManagementService.ts"; import * as ThreadTitles from "./ThreadTitleRegenerationService.ts"; +import * as WorktreeBranches from "./WorktreeBranches.ts"; it.effect.each([ ["starts the next turn after a post-capture Git status failure", false], @@ -73,6 +74,7 @@ it.effect.each([ Layer.mergeAll( projections, observer, + Layer.mock(WorktreeBranches.WorktreeBranches)({}), Layer.mock(CheckpointCapture.CheckpointCaptureServiceV2)({ execute: () => Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/RunFinalizationService.test.ts b/apps/server/src/orchestration-v2/RunFinalizationService.test.ts index 4aac21f96d29c..244276486f35b 100644 --- a/apps/server/src/orchestration-v2/RunFinalizationService.test.ts +++ b/apps/server/src/orchestration-v2/RunFinalizationService.test.ts @@ -4,9 +4,12 @@ import * as NodeFSP from "node:fs/promises"; import * as NodeOS from "node:os"; import * as NodePath from "node:path"; +import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; +import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; import { CheckpointScopeId, + GitCommandError, MessageId, ProjectId, ProviderInstanceId, @@ -18,22 +21,27 @@ import * as Deferred from "effect/Deferred"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; +import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Stream from "effect/Stream"; import * as Option from "effect/Option"; import * as TestClock from "effect/testing/TestClock"; +import * as ServerConfig from "../config.ts"; import * as CodeIntelligence from "../project/knowledge/CodeIntelligence.ts"; import * as ModelEvaluations from "../modelEvaluations/ModelEvaluations.ts"; import * as PullRequestService from "../pullRequest/PullRequestService.ts"; +import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; import * as VcsStatusBroadcaster from "../vcs/VcsStatusBroadcaster.ts"; import * as WorkspaceEntries from "../workspace/WorkspaceEntries.ts"; import * as WorkspacePaths from "../workspace/WorkspacePaths.ts"; import { withWorkspaceLease } from "../workspace/workspaceLease.ts"; import * as CheckpointCapture from "./CheckpointCaptureService.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; +import * as Orchestrator from "./Orchestrator.ts"; import * as ProjectStore from "./ProjectStore.ts"; import * as RunFinalization from "./RunFinalizationService.ts"; +import * as WorktreeBranches from "./WorktreeBranches.ts"; it.effect("coalesces completed turns without starving later project refreshes", () => { const projectA = ProjectId.make("project-turn-refresh-a"); @@ -76,7 +84,7 @@ it.effect("coalesces completed turns without starving later project refreshes", }).pipe(Effect.provide(layer)); }); -it.effect.each(["workspace refresh", "evaluation enqueue"] as const)( +it.effect.each(["workspace refresh", "evaluation enqueue", "branch sync"] as const)( "preserves completed checkpoint finalization when %s fails", (failure) => { const threadId = ThreadId.make("thread_refresh_failure"); @@ -120,6 +128,26 @@ it.effect.each(["workspace refresh", "evaluation enqueue"] as const)( checkpointScopes: [{ id: scopeId, runId, kind: "root_run" as const, cwd: "/repo" }], checkpoints: [], }), + getThreadShell: () => + Effect.succeed({ + id: threadId, + branch: "feature", + worktreePath: "/repo", + } as OrchestrationV2ThreadShell), + }), + Layer.mock(WorktreeBranches.WorktreeBranches)({ + sync: () => + Effect.gen(function* () { + captured.push("sync"); + if (failure === "branch sync") + return yield* new GitCommandError({ + operation: "test", + command: "git", + cwd: "/repo", + detail: "Git index is locked.", + }); + return { _tag: "Bound" as const, branch: "feature" }; + }), }), Layer.succeed(RunFinalization.RunFinalizationObserver, { refresh: () => @@ -151,7 +179,7 @@ it.effect.each(["workspace refresh", "evaluation enqueue"] as const)( return Effect.gen(function* () { const service = yield* RunFinalization.RunFinalizationService; yield* service.finalize({ threadId, runId, scopeId }); - assert.deepEqual(captured, ["checkpoint", "evaluation", "refresh"]); + assert.deepEqual(captured, ["checkpoint", "evaluation", "sync", "refresh"]); }).pipe(Effect.provide(layer)); }, ); @@ -239,6 +267,146 @@ it.effect.each( }).pipe(Effect.provide(layer)); }); +it.live.each( + ( + [ + { + label: "rebinds a thread that ended on a new branch before its pull request refresh", + checkout: ["switch", "-c", "agent/work"], + worktreePath: "worktree", + refName: "agent/work", + branch: "agent/work", + lookups: 1, + }, + { + label: "leaves a thread that ended on a detached HEAD alone", + checkout: ["switch", "--detach"], + worktreePath: "worktree", + refName: null, + branch: "feature", + lookups: 0, + }, + { + label: "leaves a thread that ended on the default branch alone", + checkout: ["switch", "main"], + worktreePath: "worktree", + refName: "main", + branch: "feature", + lookups: 0, + }, + { + label: "leaves a thread working in the project root alone", + checkout: ["switch", "-c", "agent/work"], + worktreePath: null, + refName: "agent/work", + branch: "feature", + lookups: 0, + }, + ] as const + ).map((scenario) => [scenario.label, scenario] as const), +)("%s", ([, scenario]) => + Effect.gen(function* () { + const threadId = ThreadId.make("thread-branch-sync"); + const runId = RunId.make("completed-run"); + const scopeId = CheckpointScopeId.make("scope-branch-sync"); + const lookedUp = yield* Deferred.make(); + const fs = yield* FileSystem.FileSystem; + const git = yield* GitVcsDriver.GitVcsDriver; + const root = yield* fs.realPath( + yield* fs.makeTempDirectoryScoped({ prefix: "t3-branch-sync-" }), + ); + const repo = `${root}/repo`; + const worktree = `${root}/feature`; + yield* fs.makeDirectory(repo); + const command = (cwd: string, args: ReadonlyArray) => + git.execute({ operation: "run-finalization-test", cwd, args }); + yield* command(repo, ["init", "-b", "main"]); + yield* command(repo, ["config", "user.email", "finalize@example.test"]); + yield* command(repo, ["config", "user.name", "Finalize test"]); + yield* command(repo, ["commit", "--allow-empty", "-m", "initial"]); + yield* command(repo, ["worktree", "add", "-b", "feature", worktree]); + if (scenario.refName === "main") yield* command(repo, ["switch", "--detach"]); + yield* command(worktree, scenario.checkout); + + let thread = { + id: threadId, + branch: "feature", + worktreePath: scenario.worktreePath === null ? null : worktree, + activeRunId: null, + archivedAt: null, + createdAt: DateTime.makeUnsafe(0), + } as OrchestrationV2ThreadShell; + let lookups = 0; + const projections = Layer.mock(ProjectionStore.ProjectionStoreV2)({ + getCheckpointContext: () => + Effect.succeed({ + runs: [], + checkpointScopes: [{ id: scopeId, runId, kind: "root_run" as const, cwd: worktree }], + checkpoints: [], + }), + getThreadShell: () => Effect.sync(() => thread), + }); + const layer = RunFinalization.layer.pipe( + Layer.provideMerge(RunFinalization.layerObserver), + Layer.provide(WorktreeBranches.layer), + Layer.provide( + Layer.mergeAll( + projections, + Layer.mock(ProjectStore.ProjectStoreV2)({ + getShell: () => Effect.succeedSome({ workspaceRoot: repo }) as never, + }), + Layer.mock(ModelEvaluations.ModelEvaluations)({}), + Layer.mock(CheckpointCapture.CheckpointCaptureServiceV2)({ execute: () => Effect.void }), + Layer.mock(Orchestrator.OrchestratorV2)({ + getShellSnapshot: () => + Effect.sync(() => ({ threads: [thread], archivedThreads: [] })) as never, + dispatch: ((command: { readonly branch: string }) => + Effect.sync(() => { + thread = { ...thread, branch: command.branch }; + })) as never, + }), + Layer.mock(WorkspaceEntries.WorkspaceEntries)({ refresh: () => Effect.void }), + Layer.mock(CodeIntelligence.CodeIntelligence)({ refreshIfIndexed: () => Effect.void }), + Layer.mock(PullRequestService.PullRequestService)({ + refreshAfterTurn: () => Effect.void, + }), + Layer.mock(VcsStatusBroadcaster.VcsStatusBroadcaster)({ + refreshLocalStatus: () => + Effect.succeed({ + isRepo: true, + hasPrimaryRemote: false, + isDefaultRef: scenario.refName === "main", + refName: scenario.refName, + hasWorkingTreeChanges: false, + workingTree: { files: [], insertions: 0, deletions: 0 }, + }), + refreshPullRequestStatus: () => + Effect.suspend(() => { + lookups++; + return Deferred.succeed(lookedUp, undefined).pipe(Effect.as(null)); + }), + }), + ), + ), + ); + yield* Effect.gen(function* () { + const service = yield* RunFinalization.RunFinalizationService; + yield* service.finalize({ threadId, runId, scopeId }); + if (scenario.lookups > 0) yield* Deferred.await(lookedUp); + assert.equal(thread.branch, scenario.branch); + assert.equal(lookups, scenario.lookups); + }).pipe(Effect.provide(layer)); + }).pipe( + Effect.provide( + GitVcsDriver.layer.pipe( + Layer.provideMerge(ServerConfig.layerTest(process.cwd(), { prefix: "t3-branch-sync-" })), + Layer.provideMerge(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)), + ), + ), + Effect.scoped, + ), +); + it.effect("finalizes a run without waiting for the code index refresh its checkpoint started", () => Effect.acquireUseRelease( Effect.promise(async () => { @@ -263,6 +431,7 @@ it.effect("finalizes a run without waiting for the code index refresh its checkp Layer.provide( Layer.mergeAll( Layer.mock(ModelEvaluations.ModelEvaluations)({}), + Layer.mock(WorktreeBranches.WorktreeBranches)({}), Layer.mock(CheckpointCapture.CheckpointCaptureServiceV2)({ execute: () => Effect.void, }), diff --git a/apps/server/src/orchestration-v2/RunFinalizationService.ts b/apps/server/src/orchestration-v2/RunFinalizationService.ts index 05f1611d0dfe6..44ce992b3114e 100644 --- a/apps/server/src/orchestration-v2/RunFinalizationService.ts +++ b/apps/server/src/orchestration-v2/RunFinalizationService.ts @@ -13,6 +13,7 @@ import * as WorkspaceEntries from "../workspace/WorkspaceEntries.ts"; import * as ModelEvaluations from "../modelEvaluations/ModelEvaluations.ts"; import * as CheckpointCapture from "./CheckpointCaptureService.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; +import * as WorktreeBranches from "./WorktreeBranches.ts"; export class RunFinalizationError extends Schema.TaggedError()( "RunFinalizationError", @@ -57,6 +58,7 @@ const make = Effect.gen(function* () { const projections = yield* ProjectionStore.ProjectionStoreV2; const observer = yield* RunFinalizationObserver; const modelEvaluations = yield* ModelEvaluations.ModelEvaluations; + const worktreeBranches = yield* WorktreeBranches.WorktreeBranches; const finalize: RunFinalizationService["Service"]["finalize"] = Effect.fn( "RunFinalizationService.finalize", @@ -94,6 +96,20 @@ const make = Effect.gen(function* () { ); } if (cwd !== undefined) { + yield* projections.getThreadShell(input.threadId).pipe( + Effect.flatMap((thread) => + // An exact match excludes project-root threads and nested-repo checkpoint scopes. + thread?.worktreePath === cwd + ? worktreeBranches.sync({ worktreePath: cwd, runId: input.runId }) + : Effect.void, + ), + Effect.catchCause((cause) => + Effect.logWarning("Failed to sync worktree branch after checkpoint capture", { + ...input, + cause, + }), + ), + ); // Retrying a display refresh after the checkpoint is durable blocks the next turn. yield* observer.refresh({ cwd, threadId: input.threadId, runId: input.runId }).pipe( Effect.catchTags({ diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts index 86edeaed33b7c..38ebd7fafb6d1 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts @@ -41,6 +41,7 @@ import * as Stream from "effect/Stream"; import * as Schema from "effect/Schema"; import * as SqlClient from "effect/sql/SqlClient"; import * as TestClock from "effect/testing/TestClock"; +import { ChildProcessSpawner } from "effect/process"; import * as GitWorkflow from "../git/GitWorkflowService.ts"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; @@ -64,6 +65,7 @@ import * as ThreadLaunch from "./ThreadLaunchService.ts"; import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; import * as ThreadManagement from "./ThreadManagementService.ts"; import * as ThreadTitleRegeneration from "./ThreadTitleRegenerationService.ts"; +import * as WorktreeBranches from "./WorktreeBranches.ts"; import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts"; import * as RemoteDelegation from "../peer/RemoteDelegation.ts"; import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; @@ -158,8 +160,55 @@ function makeHarness(options: HarnessOptions = {}) { worktree: { path: "/repo-worktrees/feature", refName: input.newRefName, headSha: "abc" }, } as never)), ); - const renameBranch = vi.fn( - options.renameBranch ?? ((input) => Effect.succeed({ branch: input.newBranch })), + let checkedOutBranch = ""; + const renameBranch = vi.fn((input) => + (options.renameBranch ?? ((request) => Effect.succeed({ branch: request.newBranch })))( + input, + ).pipe( + Effect.tap((renamed) => + Effect.sync(() => { + checkedOutBranch = renamed.branch; + }), + ), + ), + ); + const branchSynced = Deferred.makeUnsafe(); + const layerWorktreeBranches = Layer.effect( + WorktreeBranches.WorktreeBranches, + Effect.map(WorktreeBranches.WorktreeBranches, (branches) => ({ + ...branches, + sync: (input: Parameters[0]) => + branches.sync(input).pipe(Effect.ensuring(Deferred.succeed(branchSynced, undefined))), + })), + ).pipe( + // Fresh: memoization would otherwise reuse the replay harness's copy, which runs real Git. + Layer.provide(Layer.fresh(WorktreeBranches.layer)), + Layer.provide(layerOrchestrator), + Layer.provide( + Layer.mock(ProjectStore.ProjectStoreV2)({ + getShell: () => Effect.succeedSome({ workspaceRoot: "/repo" }) as never, + }), + ), + Layer.provide( + Layer.mock(GitVcsDriver.GitVcsDriver)({ + execute: (input) => + Effect.succeed({ + exitCode: ChildProcessSpawner.ExitCode(0), + stdout: + input.args[0] === "symbolic-ref" + ? `refs/heads/${checkedOutBranch}\n` + : input.args.includes("--git-common-dir") + ? "/repo/.git\n" + : "index\n", + stderr: "", + stdoutTruncated: false, + stderrTruncated: false, + }), + resolvePrimaryRemoteName: () => Effect.succeed("origin"), + resolveDefaultBranchName: () => Effect.succeed("main"), + }), + ), + Layer.provide(NodeServices.layer), ); const removeWorktree = vi.fn( (_input: Parameters[0]) => @@ -238,6 +287,7 @@ function makeHarness(options: HarnessOptions = {}) { layerExternalServices, layerThreadManagement, layerReceipts, + layerWorktreeBranches, IdAllocator.layer, ), ), @@ -289,6 +339,7 @@ function makeHarness(options: HarnessOptions = {}) { removeWorktree, deleteLocalBranch, renameBranch, + branchSynced, generateBranchName, generateThreadTitle, runSetup, @@ -2271,6 +2322,61 @@ it.effect("shared intake preserves durable attachment bytes after a lost launch }).pipe(Effect.provide(Layer.mergeAll(harness.layer, layerFiles))); }); +it.effect("rebinds every thread sharing the worktree after the generated branch rename", () => + Effect.gen(function* () { + const naming = yield* Deferred.make(); + const allowNaming = yield* Deferred.make(); + const childRecorded = yield* Deferred.make(); + const childThread = ThreadId.make("thread:rename-child"); + const harness = makeHarness({ + runSetup: (input) => + (input.threadId === childThread + ? Deferred.succeed(childRecorded, undefined) + : Effect.void + ).pipe(Effect.as({ status: "no-script" as const })), + generateBranchName: () => + Deferred.succeed(naming, undefined).pipe( + Effect.andThen(Deferred.await(allowNaming)), + Effect.as({ branch: "generated-branch" }), + ), + }); + yield* Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const threads = yield* ThreadManagement.ThreadManagementService; + const parent = yield* launches.launch( + launchInput({ + command: "launch:rename-parent", + thread: "thread:rename-parent", + message: "Start", + workspace: { type: "worktree", baseRef: "main" }, + }), + ); + yield* Deferred.await(naming); + const { branch: temporaryBranch, worktreePath } = (yield* threads.getThreadProjection( + parent.threadId, + )).thread; + assert.isNotNull(temporaryBranch); + assert.isNotNull(worktreePath); + const child = yield* launches.launch( + launchInput({ + command: "launch:rename-child", + thread: childThread, + workspace: { type: "existing_worktree", worktreePath, branch: temporaryBranch }, + }), + ); + yield* Deferred.await(childRecorded); + yield* Deferred.succeed(allowNaming, undefined); + yield* Deferred.await(harness.branchSynced); + for (const threadId of [parent.threadId, child.threadId]) { + assert.equal( + (yield* threads.getThreadProjection(threadId)).thread.branch, + "generated-branch", + ); + } + }).pipe(Effect.provide(harness.layer)); + }), +); + it.effect("cancels tracked setup before provider work is released", () => Effect.gen(function* () { const entered = yield* Deferred.make(); diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.ts index 126f65de949f7..e94141fbe3c3e 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.ts @@ -55,6 +55,7 @@ import { makeProviderFailure } from "@t3tools/provider-core/server/failure"; import { randomUuidV4 } from "@t3tools/provider-core/server/randomUuid"; import { DispatchModeLimit } from "./DispatchModeLimit.ts"; import * as ThreadManagement from "./ThreadManagementService.ts"; +import * as WorktreeBranches from "./WorktreeBranches.ts"; export type ThreadLaunchWorkspaceStrategy = | { readonly type: "root"; readonly branch?: string | undefined } @@ -216,6 +217,7 @@ const make = Effect.gen(function* () { const receipts = yield* CommandReceiptStore.CommandReceiptStoreV2; const ids = yield* IdAllocator.IdAllocatorV2; const threads = yield* ThreadManagement.ThreadManagementService; + const worktreeBranches = yield* WorktreeBranches.WorktreeBranches; const managedFolders = yield* ManagedProjectFolders.ManagedProjectFolders; const preparationScope = yield* Scope.make("sequential"); const scheduledLaunches = yield* Ref.make>(new Set()); @@ -509,55 +511,10 @@ const make = Effect.gen(function* () { Effect.uninterruptible, ), ), - Effect.flatMap((renamed) => - Effect.gen(function* () { - yield* threads - .dispatch({ - type: "thread.metadata.update", - commandId: CommandId.make(`${input.commandId}:branch-rename`), - threadId, - branch: renamed.branch, - worktreePath: worktreeCwd, - expectedBranch: oldBranch, - expectedWorktreePath: worktreeCwd, - }) - .pipe( - Effect.catchCause((cause) => - Effect.logWarning("Parent thread worktree branch update failed", { - threadId, - cause, - }), - ), - ); - // Update parents first so children inherit names; reread each pass for newly updated descendants. - const visited = new Set([threadId]); - while (true) { - const snapshot = yield* threads.getShellSnapshot({ worktreePath: worktreeCwd }); - const sharing = [...snapshot.threads, ...snapshot.archivedThreads].filter( - (thread) => !visited.has(thread.id) && thread.branch === oldBranch, - ); - if (sharing.length === 0) break; - for (const thread of sharing) { - visited.add(thread.id); - yield* threads - .dispatch({ - type: "thread.metadata.update", - commandId: CommandId.make(`${input.commandId}:branch-rename:${thread.id}`), - threadId: thread.id, - branch: renamed.branch, - expectedBranch: oldBranch, - expectedWorktreePath: worktreeCwd, - }) - .pipe( - Effect.catchCause((cause) => - Effect.logWarning("Shared thread worktree branch update failed", { - threadId: thread.id, - cause, - }), - ), - ); - } - } + Effect.flatMap(() => + worktreeBranches.sync({ + worktreePath: worktreeCwd, + ...(runId === null ? {} : { runId }), }), ), Effect.catchCause((cause) => diff --git a/apps/server/src/orchestration-v2/WorktreeBranches.test.ts b/apps/server/src/orchestration-v2/WorktreeBranches.test.ts new file mode 100644 index 0000000000000..0e11cddef8ca3 --- /dev/null +++ b/apps/server/src/orchestration-v2/WorktreeBranches.test.ts @@ -0,0 +1,454 @@ +import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import { + RunId, + ThreadId, + type CommandId, + type OrchestrationV2ThreadShell, +} from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import type * as Path from "effect/Path"; +import type * as Scope from "effect/Scope"; + +import * as ServerConfig from "../config.ts"; +import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; +import * as Orchestrator from "./Orchestrator.ts"; +import * as ProjectStore from "./ProjectStore.ts"; +import * as WorktreeBranches from "./WorktreeBranches.ts"; + +interface MetadataUpdate { + readonly commandId: CommandId; + readonly threadId: ThreadId; + readonly branch: string; + readonly expectedBranch: string | null; + readonly expectedWorktreePath: string; +} + +const gitLayer = GitVcsDriver.layer.pipe( + Layer.provideMerge(ServerConfig.layerTest(process.cwd(), { prefix: "worktree-branches-" })), + Layer.provideMerge(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)), +); + +const member = ( + id: string, + branch: string | null, + worktreePath: string, + overrides: Partial = {}, +) => + ({ + id: ThreadId.make(id), + branch, + worktreePath, + activeRunId: null, + archivedAt: null, + createdAt: DateTime.makeUnsafe(0), + ...overrides, + }) as OrchestrationV2ThreadShell; + +const fixture = Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const git = yield* GitVcsDriver.GitVcsDriver; + const root = yield* fs.realPath(yield* fs.makeTempDirectoryScoped({ prefix: "worktree-sync-" })); + const repo = `${root}/repo`; + const worktree = `${root}/feature`; + yield* fs.makeDirectory(repo); + const command = (cwd: string, args: string[]) => + git + .execute({ operation: "worktree-branches-test", cwd, args }) + .pipe(Effect.map((result) => result.stdout.trim())); + yield* command(repo, ["init", "-b", "main"]); + yield* command(repo, ["config", "user.email", "branches@example.test"]); + yield* command(repo, ["config", "user.name", "Branches test"]); + yield* command(repo, ["commit", "--allow-empty", "-m", "initial"]); + yield* command(repo, ["worktree", "add", "-b", "feature", worktree]); + + let threads: OrchestrationV2ThreadShell[] = []; + let onDispatch: (update: MetadataUpdate) => Effect.Effect = () => Effect.void; + const dispatched: MetadataUpdate[] = []; + const context = yield* Layer.build(WorktreeBranches.layer).pipe( + Effect.provideService(ProjectStore.ProjectStoreV2, { + getShell: () => Effect.succeedSome({ workspaceRoot: repo }), + } as unknown as ProjectStore.ProjectStoreV2["Service"]), + Effect.provideService(Orchestrator.OrchestratorV2, { + getShellSnapshot: (options: { readonly worktreePath: string }) => + Effect.sync(() => { + const members = threads.filter((thread) => thread.worktreePath === options.worktreePath); + return { + threads: members.filter((thread) => thread.archivedAt === null), + archivedThreads: members.filter((thread) => thread.archivedAt !== null), + }; + }), + dispatch: (update: MetadataUpdate) => + Effect.gen(function* () { + yield* onDispatch(update); + dispatched.push(update); + threads = threads.map((thread) => + thread.id === update.threadId && + thread.branch === update.expectedBranch && + thread.worktreePath === update.expectedWorktreePath + ? { ...thread, branch: update.branch } + : thread, + ); + }), + } as unknown as Orchestrator.OrchestratorV2["Service"]), + ); + return { + service: Context.get(context, WorktreeBranches.WorktreeBranches), + command, + repo, + worktree, + dispatched, + threads: () => threads, + setThreads: (next: OrchestrationV2ThreadShell[]) => { + threads = next; + }, + onDispatch: (hook: (update: MetadataUpdate) => Effect.Effect) => { + onDispatch = hook; + }, + }; +}); + +const run = ( + effect: Effect.Effect< + A, + E, + | Scope.Scope + | FileSystem.FileSystem + | GitVcsDriver.GitVcsDriver + | Path.Path + | ServerConfig.ServerConfig + >, +) => effect.pipe(Effect.provide(gitLayer), Effect.scoped); + +it.live("returns Bound without writing when every thread matches HEAD", () => + run( + Effect.gen(function* () { + const { service, worktree, dispatched, setThreads } = yield* fixture; + setThreads([member("thread", "feature", worktree)]); + assert.deepEqual(yield* service.sync({ worktreePath: worktree }), { + _tag: "Bound", + branch: "feature", + }); + assert.deepEqual(dispatched, []); + }), + ), +); + +it.live("writes every member oldest first, each with its own swap, and a repeat is Bound", () => + run( + Effect.gen(function* () { + const { service, command, worktree, dispatched, threads, setThreads } = yield* fixture; + yield* command(worktree, ["switch", "-c", "agent/work"]); + setThreads([ + member("child", "t3/old", worktree, { createdAt: DateTime.makeUnsafe(2) }), + member("no-branch", null, worktree, { createdAt: DateTime.makeUnsafe(3) }), + member("archived", "feature", worktree, { + createdAt: DateTime.makeUnsafe(4), + archivedAt: DateTime.makeUnsafe(5), + }), + member("parent", "feature", worktree, { createdAt: DateTime.makeUnsafe(1) }), + member("elsewhere", "feature", `${worktree}-other`), + ]); + assert.deepEqual(yield* service.sync({ worktreePath: worktree }), { + _tag: "Rebinding", + branch: "agent/work", + }); + assert.deepEqual( + dispatched.map((update) => [update.threadId, update.expectedBranch]), + [ + ["parent", "feature"], + ["child", "t3/old"], + ["no-branch", null], + ["archived", "feature"], + ], + ); + assert.isTrue(dispatched.every((update) => update.expectedWorktreePath === worktree)); + assert.equal(new Set(dispatched.map((update) => update.commandId)).size, 4); + assert.equal(threads().find((thread) => thread.id === "elsewhere")?.branch, "feature"); + assert.deepEqual(yield* service.sync({ worktreePath: worktree }), { + _tag: "Bound", + branch: "agent/work", + }); + }), + ), +); + +it.live.each([ + ["a detached HEAD", ["switch", "--detach"], "Detached", undefined], + ["an unborn branch", ["switch", "--orphan", "unborn"], "Unavailable", undefined], + ["a parenthesized name", ["switch", "-c", "(feature)"], "Unavailable", undefined], + ["the default branch", ["switch", "main"], "Unbound", "main"], +] as const)("writes nothing for %s", ([, checkout, tag, branch]) => + run( + Effect.gen(function* () { + const { service, command, repo, worktree, dispatched, setThreads } = yield* fixture; + setThreads([member("thread", "feature", worktree)]); + if (branch === "main") yield* command(repo, ["switch", "--detach"]); + yield* command(worktree, [...checkout]); + assert.deepEqual(yield* service.sync({ worktreePath: worktree }), { + _tag: tag, + ...(branch === undefined ? {} : { branch }), + } as WorktreeBranches.WorktreeCheckout); + assert.deepEqual(dispatched, []); + }), + ), +); + +it.live("writes nothing for a branch another checkout holds", () => + run( + Effect.gen(function* () { + const { service, command, repo, worktree, dispatched, setThreads } = yield* fixture; + setThreads([member("thread", "feature", worktree)]); + yield* command(repo, ["switch", "-c", "trunk"]); + yield* command(worktree, ["switch", "--ignore-other-worktrees", "trunk"]); + assert.deepEqual(yield* service.sync({ worktreePath: worktree }), { + _tag: "Unbound", + branch: "trunk", + }); + assert.deepEqual(dispatched, []); + }), + ), +); + +it.live("writes nothing when another repository replaced the worktree", () => + run( + Effect.gen(function* () { + const { service, command, worktree, dispatched, setThreads } = yield* fixture; + const fs = yield* FileSystem.FileSystem; + const other = yield* fs.makeTempDirectoryScoped({ prefix: "worktree-sync-other-" }); + yield* command(other, ["init", "-b", "main"]); + yield* command(other, ["config", "user.email", "other@example.test"]); + yield* command(other, ["config", "user.name", "Other"]); + yield* command(other, ["commit", "--allow-empty", "-m", "unrelated"]); + yield* fs.rename(worktree, `${worktree}-original`); + yield* command(other, ["worktree", "add", "-b", "other-work", worktree]); + setThreads([member("thread", "feature", worktree)]); + assert.deepEqual(yield* service.sync({ worktreePath: worktree }), { + _tag: "Unavailable", + }); + assert.deepEqual(dispatched, []); + }), + ), +); + +it.live("reads the default branch from the primary remote before main or master", () => + run( + Effect.gen(function* () { + const { service, command, repo, worktree, dispatched, setThreads } = yield* fixture; + setThreads([member("thread", "feature", worktree)]); + yield* command(repo, ["branch", "trunk", "main"]); + yield* command(repo, ["remote", "add", "origin", repo]); + yield* command(repo, ["fetch", "origin"]); + yield* command(repo, [ + "symbolic-ref", + "refs/remotes/origin/HEAD", + "refs/remotes/origin/trunk", + ]); + yield* command(worktree, ["switch", "trunk"]); + assert.equal((yield* service.sync({ worktreePath: worktree }))._tag, "Unbound"); + assert.deepEqual(dispatched, []); + yield* command(repo, ["switch", "--detach"]); + yield* command(worktree, ["switch", "main"]); + assert.equal((yield* service.sync({ worktreePath: worktree }))._tag, "Rebinding"); + }), + ), +); + +it.live("asks the remote when no local HEAD ref says whether main is its default", () => + run( + Effect.gen(function* () { + const { service, command, repo, worktree, dispatched, setThreads } = yield* fixture; + const fs = yield* FileSystem.FileSystem; + const remote = yield* fs.makeTempDirectoryScoped({ prefix: "worktree-sync-remote-" }); + setThreads([member("thread", "feature", worktree)]); + yield* command(remote, ["init", "--bare", "-b", "trunk"]); + yield* command(repo, ["branch", "-m", "trunk"]); + yield* command(repo, ["remote", "add", "origin", remote]); + yield* command(repo, ["push", "origin", "trunk"]); + yield* command(worktree, ["switch", "-c", "main"]); + assert.equal((yield* service.sync({ worktreePath: worktree }))._tag, "Rebinding"); + assert.deepEqual( + dispatched.map((update) => update.branch), + ["main"], + ); + }), + ), +); + +it.live("treats the remote's default of any name as default, and asks once per checkout", () => + run( + Effect.gen(function* () { + const { service, command, repo, worktree, dispatched, setThreads } = yield* fixture; + const fs = yield* FileSystem.FileSystem; + const remote = yield* fs.makeTempDirectoryScoped({ prefix: "worktree-sync-remote-" }); + setThreads([member("thread", "feature", worktree)]); + yield* command(remote, ["init", "--bare", "-b", "develop"]); + yield* command(repo, ["branch", "-m", "develop"]); + yield* command(repo, ["remote", "add", "origin", remote]); + yield* command(repo, ["push", "origin", "develop"]); + yield* command(repo, ["switch", "--detach"]); + yield* command(worktree, ["switch", "develop"]); + assert.equal((yield* service.sync({ worktreePath: worktree }))._tag, "Unbound"); + // Offline, a second question would fall back to main and master and call this a new branch. + yield* fs.remove(remote, { recursive: true }); + assert.equal((yield* service.sync({ worktreePath: worktree }))._tag, "Unbound"); + assert.deepEqual(dispatched, []); + }), + ), +); + +it.live("recognizes the same repository however its path is spelled", () => + run( + Effect.gen(function* () { + const { service, command, repo, worktree } = yield* fixture; + const fs = yield* FileSystem.FileSystem; + assert.isTrue(yield* service.sameRepository(worktree, repo)); + const other = yield* fs.makeTempDirectoryScoped({ prefix: "worktree-sync-other-" }); + yield* command(other, ["init", "-b", "main"]); + assert.isFalse(yield* service.sameRepository(worktree, other)); + assert.isFalse(yield* service.sameRepository(worktree, `${repo}-missing`)); + // Only a case-insensitive disk accepts the other spelling at all. + const upper = repo.toUpperCase(); + if (yield* fs.exists(upper)) assert.isTrue(yield* service.sameRepository(worktree, upper)); + }), + ), +); + +it.live("reports a missing worktree as Unavailable", () => + run( + Effect.gen(function* () { + const { service, worktree } = yield* fixture; + const missing = `${worktree}-missing`; + assert.deepEqual(yield* service.observe(missing), { _tag: "Unavailable" }); + assert.deepEqual(yield* service.sync({ worktreePath: missing }), { + _tag: "Unavailable", + }); + }), + ), +); + +it.live("defers to a sibling's active run but writes through the caller's own", () => + run( + Effect.gen(function* () { + const { service, command, worktree, dispatched, setThreads } = yield* fixture; + const runId = RunId.make("run-own"); + yield* command(worktree, ["switch", "-c", "agent/work"]); + setThreads([ + member("own", "feature", worktree, { activeRunId: runId }), + member("sibling", "feature", worktree, { activeRunId: RunId.make("run-sibling") }), + ]); + const sync = { worktreePath: worktree, runId }; + assert.deepEqual(yield* service.sync(sync), { _tag: "Rebinding", branch: "agent/work" }); + assert.deepEqual(dispatched, []); + setThreads([ + member("own", "feature", worktree, { activeRunId: runId }), + member("sibling", "feature", worktree), + ]); + yield* service.sync(sync); + assert.deepEqual(dispatched.map((update) => update.threadId).toSorted(), ["own", "sibling"]); + }), + ), +); + +it.live("logs a lost swap and still writes the next member", () => + run( + Effect.gen(function* () { + const { service, command, worktree, threads, setThreads, onDispatch } = yield* fixture; + yield* command(worktree, ["switch", "-c", "agent/work"]); + setThreads([ + member("first", "feature", worktree, { createdAt: DateTime.makeUnsafe(1) }), + member("second", "feature", worktree, { createdAt: DateTime.makeUnsafe(2) }), + ]); + onDispatch((update) => (update.threadId === "first" ? Effect.die("swap lost") : Effect.void)); + assert.equal((yield* service.sync({ worktreePath: worktree }))._tag, "Rebinding"); + assert.deepEqual( + threads().map((thread) => thread.branch), + ["feature", "agent/work"], + ); + }), + ), +); + +it.live("picks up a child created during the write when it rereads", () => + run( + Effect.gen(function* () { + const { service, command, worktree, threads, setThreads, onDispatch } = yield* fixture; + yield* command(worktree, ["switch", "-c", "agent/work"]); + setThreads([member("parent", "feature", worktree)]); + onDispatch((update) => + Effect.sync(() => { + if (update.threadId === "parent") + setThreads([...threads(), member("child", "feature", worktree)]); + }), + ); + yield* service.sync({ worktreePath: worktree }); + assert.deepEqual( + threads().map((thread) => thread.branch), + ["agent/work", "agent/work"], + ); + }), + ), +); + +it.live("terminates when every write is rejected", () => + run( + Effect.gen(function* () { + const { service, command, worktree, setThreads, onDispatch } = yield* fixture; + yield* command(worktree, ["switch", "-c", "agent/work"]); + setThreads([member("parent", "feature", worktree)]); + onDispatch(() => Effect.die("rejected")); + assert.equal((yield* service.sync({ worktreePath: worktree }))._tag, "Rebinding"); + }), + ), +); + +it.live("corrects a sibling whose write was deferred when another run ends", () => + run( + Effect.gen(function* () { + const { service, command, worktree, threads, setThreads } = yield* fixture; + yield* command(worktree, ["branch", "-m", "feature", "t3/named"]); + const childRun = RunId.make("run-child"); + setThreads([ + member("parent", "t3/temp", worktree), + member("child", "t3/named", worktree, { activeRunId: childRun }), + ]); + const parentEnd = yield* service.sync({ + worktreePath: worktree, + runId: RunId.make("run-parent"), + }); + assert.equal(parentEnd._tag, "Rebinding"); + assert.equal(threads().find((thread) => thread.id === "parent")?.branch, "t3/temp"); + yield* service.sync({ worktreePath: worktree, runId: childRun }); + assert.equal(threads().find((thread) => thread.id === "parent")?.branch, "t3/named"); + }), + ), +); + +it.live("rebinds while a Git command holds the index lock, which observe reports", () => + run( + Effect.gen(function* () { + const { service, command, worktree, dispatched, setThreads } = yield* fixture; + const fs = yield* FileSystem.FileSystem; + yield* command(worktree, ["switch", "-c", "agent/work"]); + setThreads([member("own", "feature", worktree)]); + const index = yield* command(worktree, [ + "rev-parse", + "--path-format=absolute", + "--git-path", + "index", + ]); + yield* fs.writeFileString(`${index}.lock`, ""); + assert.equal((yield* service.sync({ worktreePath: worktree }))._tag, "Rebinding"); + assert.deepEqual( + dispatched.map((update) => update.threadId), + ["own"], + ); + assert.equal((yield* Effect.exit(service.observe(worktree)))._tag, "Failure"); + }), + ), +); diff --git a/apps/server/src/orchestration-v2/WorktreeBranches.ts b/apps/server/src/orchestration-v2/WorktreeBranches.ts new file mode 100644 index 0000000000000..214cf03c3d9da --- /dev/null +++ b/apps/server/src/orchestration-v2/WorktreeBranches.ts @@ -0,0 +1,312 @@ +import { CommandId, GitCommandError, type RunId } from "@t3tools/contracts"; +import { randomUuidV4 } from "@t3tools/provider-core/server/randomUuid"; +import * as Context from "effect/Context"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import type { PlatformError } from "effect/PlatformError"; + +import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; +import * as Orchestrator from "./Orchestrator.ts"; +import * as ProjectStore from "./ProjectStore.ts"; + +const REMOTE_HEAD_TIMEOUT_MS = 5_000; + +export type WorktreeHead = + | { readonly _tag: "Unavailable" } + | { readonly _tag: "Detached" } + | { readonly _tag: "Named"; readonly branch: string }; + +export type WorktreeCheckout = + | { readonly _tag: "Unavailable" } + | { readonly _tag: "Detached" } + /** Records differ, but a thread could not recreate its worktree on this branch: the default branch or one another checkout holds. Its own ref holds HEAD, so nothing is written. */ + | { readonly _tag: "Unbound"; readonly branch: string } + | { readonly _tag: "Bound"; readonly branch: string } + /** Records differ, and a write was attempted or deferred. The caller's thread data is stale. */ + | { readonly _tag: "Rebinding"; readonly branch: string }; + +type WorktreeBranchError = GitCommandError | PlatformError; + +export class WorktreeBranches extends Context.Service< + WorktreeBranches, + { + readonly observe: (worktreePath: string) => Effect.Effect; + /** + * Rebinds the worktree's threads to its checked-out branch. `runId` names the one run allowed + * to stay active during the write. `threadPaths` lists how those threads store the path, when + * that can differ from `worktreePath`. + */ + readonly sync: (input: { + readonly worktreePath: string; + readonly threadPaths?: ReadonlyArray; + readonly runId?: RunId; + }) => Effect.Effect; + /** False when either side cannot be read, so a caller about to delete keeps what it found. */ + readonly sameRepository: (worktreePath: string, projectRoot: string) => Effect.Effect; + } +>()("t3/orchestration-v2/WorktreeBranches") {} + +const make = Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projects = yield* ProjectStore.ProjectStoreV2; + const git = yield* GitVcsDriver.GitVcsDriver; + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + + const failure = (cwd: string, detail: string) => + new GitCommandError({ + operation: "WorktreeBranches.observe", + command: "git", + cwd, + detail, + }); + + // Return null for missing/non-repository cwd; force untranslated git diagnostics for matching. + const read = Effect.fnUntraced(function* (cwd: string, args: string[], allowedExitCode?: number) { + const result = yield* git + .execute({ + operation: "WorktreeBranches.observe", + cwd, + args, + env: { LC_ALL: "C" }, + allowNonZeroExit: true, + }) + .pipe( + Effect.catch((error) => + Effect.flatMap(fs.exists(cwd), (exists) => + exists ? Effect.fail(error) : Effect.succeed(null), + ), + ), + ); + if (result === null) return null; + if (result.exitCode === 0 || result.exitCode === allowedExitCode) return result; + const stderr = result.stderr.toLowerCase(); + // Newer Git reports a dangling .git file as "gitfile does not point to a valid repository". + if ( + stderr.includes("not a git repository") || + stderr.includes("does not point to a valid repository") + ) + return null; + return yield* failure( + cwd, + result.stderr.trim().split(/\r?\n/)[0]?.slice(0, 1000) || "command failed", + ); + }); + + const readHead = Effect.fnUntraced(function* (cwd: string) { + // Detached HEAD exits 1; full refs avoid shortening when a tag shares the branch name. + const head = yield* read(cwd, ["symbolic-ref", "--quiet", "HEAD"], 1); + if (head === null) return { _tag: "Unavailable" } as const; + if (head.exitCode !== 0) return { _tag: "Detached" } as const; + const ref = head.stdout.trim(); + const branch = ref.startsWith("refs/heads/") ? ref.slice("refs/heads/".length).trim() : null; + return branch === null || branch.startsWith("(") + ? ({ _tag: "Unavailable" } as const) + : ({ _tag: "Named", branch } as const); + }); + + const observe: WorktreeBranches["Service"]["observe"] = Effect.fn("WorktreeBranches.observe")( + function* (cwd) { + const index = yield* read(cwd, ["rev-parse", "--git-path", "index"]); + if (index === null) return { _tag: "Unavailable" }; + if (yield* fs.exists(`${path.resolve(cwd, index.stdout.trim())}.lock`)) + return yield* failure( + cwd, + "Git index is locked. Status will resume when the index lock is removed.", + ); + return yield* readHead(cwd); + }, + ); + + const readMembers = Effect.fnUntraced(function* (paths: ReadonlyArray) { + const snapshots = yield* Effect.forEach(paths, (worktreePath) => + orchestrator.getShellSnapshot({ worktreePath }), + ); + return snapshots.flatMap((snapshot) => [...snapshot.threads, ...snapshot.archivedThreads]); + }); + + const repositoryIdentity = Effect.fnUntraced(function* (cwd: string) { + const physical = (path: string) => fs.realPath(path).pipe(Effect.orElseSucceed(() => path)); + const { stdout } = yield* git.execute({ + operation: "WorktreeBranches.sync", + cwd, + args: ["rev-parse", "--git-common-dir"], + }); + // Git prints this relative to the physical cwd, which can differ from a symlink. + const commonDir = yield* physical(path.resolve(yield* physical(cwd), stdout.trim())); + // By file, not by spelling: a case-insensitive disk accepts one folder under several names. + return yield* fs.stat(commonDir).pipe( + Effect.map(({ dev, ino }) => + Option.match(ino, { + onNone: () => commonDir, + onSome: (value) => `${dev}:${value}`, + }), + ), + Effect.orElseSucceed(() => commonDir), + ); + }); + + const sameRepository: WorktreeBranches["Service"]["sameRepository"] = Effect.fn( + "WorktreeBranches.sameRepository", + )( + function* (worktreePath, projectRoot) { + return (yield* repositoryIdentity(worktreePath)) === (yield* repositoryIdentity(projectRoot)); + }, + Effect.orElseSucceed(() => false), + ); + + // A replaced folder looks like a switched branch, but its branch means nothing to these threads. + const inProjectRepository = Effect.fn("WorktreeBranches.inProjectRepository")( + function* (worktreePath: string, paths: ReadonlyArray) { + const [member] = yield* readMembers(paths); + if (member === undefined) return true; + const project = yield* projects.getShell(member.projectId); + if (Option.isNone(project)) return false; + return yield* sameRepository(worktreePath, project.value.workspaceRoot); + }, + Effect.orElseSucceed(() => false), + ); + + // Once per checkout: a thread left on the default branch syncs at the end of every turn. + const remoteDefaults = new Map(); + + const isDefaultBranch = Effect.fn("WorktreeBranches.isDefaultBranch")(function* ( + cwd: string, + branch: string, + ) { + const remote = yield* git.resolvePrimaryRemoteName(cwd).pipe(Effect.orElseSucceed(() => null)); + if (remote === null) return GitVcsDriver.isDefaultBranchName(branch, null); + const known = yield* git.resolveDefaultBranchName(cwd, remote); + if (known !== null) return GitVcsDriver.isDefaultBranchName(branch, known); + // Without a local HEAD ref, only the remote can name its default. + const key = `${cwd}\0${remote}`; + let remoteHead = remoteDefaults.get(key); + if (remoteHead === undefined) { + remoteHead = yield* git + .execute({ + operation: "WorktreeBranches.isDefaultBranch", + cwd, + args: ["ls-remote", "--symref", "--", remote, "HEAD"], + timeoutMs: REMOTE_HEAD_TIMEOUT_MS, + }) + .pipe( + Effect.map((result) => GitVcsDriver.parseRemoteHeadBranch(result.stdout)), + Effect.orElseSucceed(() => null), + ); + remoteDefaults.set(key, remoteHead); + } + return GitVcsDriver.isDefaultBranchName(branch, remoteHead); + }); + + // Git refuses to recreate a worktree on a branch another checkout holds. + const heldByAnotherCheckout = Effect.fn("WorktreeBranches.heldByAnotherCheckout")(function* ( + cwd: string, + branch: string, + ) { + const { stdout } = yield* git.execute({ + operation: "WorktreeBranches.sync", + cwd, + args: ["worktree", "list", "--porcelain"], + }); + const holders = stdout + .split("\n") + .filter((line) => line.trim() === `branch refs/heads/${branch}`); + return holders.length > 1; + }); + + const writeMembers = Effect.fn("WorktreeBranches.writeMembers")(function* (input: { + readonly paths: ReadonlyArray; + readonly branch: string; + readonly runId: RunId | undefined; + }) { + const { paths, branch, runId } = input; + // Oldest first so parents are written before children; reread for threads created mid-write. + const visited = new Set(); + while (true) { + const members = yield* readMembers(paths); + // A sibling's agent may be mid-checkout, so HEAD proves nothing yet. Its run end syncs. + if (members.some((member) => member.activeRunId !== null && member.activeRunId !== runId)) + return; + const pending = members + .filter((member) => !visited.has(member.id) && member.branch !== branch) + .toSorted( + (left, right) => + DateTime.toEpochMillis(left.createdAt) - DateTime.toEpochMillis(right.createdAt), + ); + if (pending.length === 0) return; + for (const member of pending) { + visited.add(member.id); + // Random ids: a deterministic receipt would block a later A to B to A to B sequence. + const uuid = yield* randomUuidV4; + yield* orchestrator + .dispatch({ + type: "thread.metadata.update", + commandId: CommandId.make(`server:worktree-branch:${member.id}:${uuid}`), + threadId: member.id, + branch, + expectedBranch: member.branch, + expectedWorktreePath: member.worktreePath, + }) + .pipe( + Effect.catchCause((cause) => + Effect.logWarning("worktree branch update failed", { + threadId: member.id, + branch, + cause, + }), + ), + ); + } + } + }); + + const sync: WorktreeBranches["Service"]["sync"] = Effect.fn("WorktreeBranches.sync")(function* ({ + worktreePath, + threadPaths, + runId, + }) { + // Not `observe`: a held index lock says nothing about which branch is checked out. + const head = yield* readHead(worktreePath); + if (head._tag !== "Named") return head; + const { branch } = head; + const paths = threadPaths ?? [worktreePath]; + // Every thread, not only the caller's: a sibling's write may have been deferred to this sync. + const bound = yield* readMembers(paths).pipe( + Effect.map((members) => members.every((member) => member.branch === branch)), + // The write below reads again and logs a read that keeps failing. + Effect.orElseSucceed(() => false), + ); + if (bound) return { _tag: "Bound", branch }; + const born = yield* git.execute({ + operation: "WorktreeBranches.sync", + cwd: worktreePath, + args: ["rev-parse", "--verify", "--quiet", `refs/heads/${branch}`], + allowNonZeroExit: true, + }); + if (born.exitCode !== 0) return { _tag: "Unavailable" }; + if (!(yield* inProjectRepository(worktreePath, paths))) return { _tag: "Unavailable" }; + if ( + (yield* isDefaultBranch(worktreePath, branch)) || + (yield* heldByAnotherCheckout(worktreePath, branch)) + ) + return { _tag: "Unbound", branch }; + yield* writeMembers({ paths, branch, runId }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("worktree branch sync could not read its threads", { + worktreePath, + branch, + cause, + }), + ), + ); + return { _tag: "Rebinding", branch }; + }); + + return WorktreeBranches.of({ observe, sync, sameRepository }); +}); + +export const layer = Layer.effect(WorktreeBranches, make); diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index af46070c626e8..8615346dd21c4 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -58,6 +58,7 @@ import * as ThreadManagementService from "./ThreadManagementService.ts"; import * as ThreadLaunchService from "./ThreadLaunchService.ts"; import * as ThreadForkService from "./ThreadForkService.ts"; import * as TurnItemPositionStore from "./TurnItemPositionStore.ts"; +import * as WorktreeBranches from "./WorktreeBranches.ts"; import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts"; import * as SecretRequests from "../secrets/SecretRequests.ts"; @@ -185,10 +186,6 @@ const layerCheckpointCaptureServiceProvided = CheckpointCaptureService.layer.pip ), ), ); -const layerRunFinalizationServiceProvided = RunFinalizationService.layer.pipe( - Layer.provide(Layer.merge(layerCheckpointCaptureServiceProvided, ProjectionStore.layer)), -); - const layerOrchestratorProvided = Orchestrator.layer.pipe( Layer.provide( Layer.mergeAll( @@ -212,6 +209,20 @@ const layerOrchestratorProvided = Orchestrator.layer.pipe( ), ); +const layerWorktreeBranchesProvided = WorktreeBranches.layer.pipe( + Layer.provide(Layer.mergeAll(layerOrchestratorProvided, GitVcsDriver.layer, ProjectStore.layer)), +); + +const layerRunFinalizationServiceProvided = RunFinalizationService.layer.pipe( + Layer.provide( + Layer.mergeAll( + layerCheckpointCaptureServiceProvided, + ProjectionStore.layer, + layerWorktreeBranchesProvided, + ), + ), +); + // The usage-limit hook dispatches into the orchestrator, so run execution is // built after it; the orchestrator itself never calls run execution. const layerRunExecutionServiceProvided = RunExecutionService.layer.pipe( @@ -285,6 +296,7 @@ const layerThreadLaunchProvided = ThreadLaunchService.layer.pipe( layerManagedProjectFoldersProvided, layerThreadManagementProvided, layerCommandReceiptStoreProvided, + layerWorktreeBranchesProvided, IdAllocator.layer, ), ), diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index de3e0e91b745d..3e954cf9d694a 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -62,6 +62,7 @@ import * as McpAppModelContext from "../../mcpApps/McpAppModelContext.ts"; import * as PreviewAgentActivity from "../../preview/PreviewAgentActivity.ts"; import * as RunExecutionService from "../RunExecutionService.ts"; import * as RunFinalizationService from "../RunFinalizationService.ts"; +import * as WorktreeBranches from "../WorktreeBranches.ts"; import * as ThreadTitleRegenerationService from "../ThreadTitleRegenerationService.ts"; import * as RuntimePolicy from "../RuntimePolicy.ts"; import * as TurnItemPositionStore from "../TurnItemPositionStore.ts"; @@ -350,6 +351,10 @@ export function layerWithRegistry( const layerCommandReceiptStoreProvided = CommandReceiptStore.layer.pipe( Layer.provide(layerDatabase), ); + const layerGitVcsDriver = GitVcsDriver.layer.pipe( + Layer.provide(layerServerConfig), + Layer.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)), + ); const layerProviderEventIngestorProvided = ProviderEventIngestor.layer.pipe( Layer.provide( Layer.mergeAll( @@ -360,10 +365,7 @@ export function layerWithRegistry( layerEventSinkProvided, IdAllocator.layer, ThreadCommandExecutor.layer, - GitVcsDriver.layer.pipe( - Layer.provide(layerServerConfig), - Layer.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)), - ), + layerGitVcsDriver, ), ), ); @@ -448,9 +450,6 @@ export function layerWithRegistry( ), ), ); - const layerRunFinalizationServiceProvided = RunFinalizationService.layer.pipe( - Layer.provide(Layer.merge(layerCheckpointCaptureServiceProvided, layerStores)), - ); const layerThreadTitleRegenerationTest = Layer.succeed( ThreadTitleRegenerationService.ThreadTitleRegenerationService, ThreadTitleRegenerationService.ThreadTitleRegenerationService.of({ @@ -477,6 +476,17 @@ export function layerWithRegistry( ), ), ); + const layerRunFinalizationServiceProvided = RunFinalizationService.layer.pipe( + Layer.provide( + Layer.mergeAll( + layerCheckpointCaptureServiceProvided, + layerStores, + WorktreeBranches.layer.pipe( + Layer.provide(Layer.mergeAll(layerOrchestratorProvided, layerGitVcsDriver, layerStores)), + ), + ), + ), + ); const { poolFailoverDispatch } = options; const layerFailoverOrchestrator = poolFailoverDispatch === undefined diff --git a/apps/server/src/pstack/pstackModels.test.ts b/apps/server/src/pstack/pstackModels.test.ts index c0b6a7e4bb5b5..8939f67a2616f 100644 --- a/apps/server/src/pstack/pstackModels.test.ts +++ b/apps/server/src/pstack/pstackModels.test.ts @@ -89,7 +89,10 @@ it.effect("leaves speed to the provider unless the models file sets it", () => { providerInstanceId: "claudeAgent", model: "claude-opus-5-5", - options: [{ id: "effort", value: "xhigh" }], + options: [ + { id: "effort", value: "xhigh" }, + { id: "ultracode", value: false }, + ], }, { providerInstanceId: "codex", diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 20d0ca795dcb5..74b917bdce8ed 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -219,6 +219,7 @@ import * as ThreadSettlementService from "./orchestration-v2/ThreadSettlementSer import * as IdleCompactionService from "./orchestration-v2/IdleCompactionService.ts"; import * as ThreadPullRequestService from "./orchestration-v2/ThreadPullRequestService.ts"; import * as RunFinalizationService from "./orchestration-v2/RunFinalizationService.ts"; +import * as WorktreeBranches from "./orchestration-v2/WorktreeBranches.ts"; import * as EffectOutboxV2 from "./orchestration-v2/EffectOutbox.ts"; import * as ProjectionMaintenance from "./orchestration-v2/ProjectionMaintenance.ts"; import * as ProjectionStoreV2 from "./orchestration-v2/ProjectionStore.ts"; @@ -661,7 +662,9 @@ const layerRuntimeCoreDependenciesBase = Layer.mergeAll( HeldHooksWaker.layer, layerThreadSettlementWorker, layerIdleCompactionWorker, - StorageCleanup.layer.pipe(Layer.provide(ProjectionStoreV2.layer)), + StorageCleanup.layer.pipe( + Layer.provide(Layer.merge(ProjectionStoreV2.layer, WorktreeBranches.layer)), + ), DatabaseStorage.layer.pipe(Layer.provide(ProjectionMaintenance.layer)), layerThreadPullRequestWorker, DelegatedTaskAnalytics.layer, diff --git a/apps/server/src/storageCleanup.test.ts b/apps/server/src/storageCleanup.test.ts index b4c9ac3e68945..ad4f7b74e9b15 100644 --- a/apps/server/src/storageCleanup.test.ts +++ b/apps/server/src/storageCleanup.test.ts @@ -1,5 +1,6 @@ import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import * as NodeServices from "@effect/platform-node/NodeServices"; +import * as Clock from "effect/Clock"; import * as Effect from "effect/Effect"; import * as Context from "effect/Context"; import * as Deferred from "effect/Deferred"; @@ -30,6 +31,7 @@ import * as ProjectionStore from "./orchestration-v2/ProjectionStore.ts"; import * as ProviderSessionManager from "./orchestration-v2/ProviderSessionManager.ts"; import * as ProjectStore from "./orchestration-v2/ProjectStore.ts"; import * as Orchestrator from "./orchestration-v2/Orchestrator.ts"; +import * as WorktreeBranches from "./orchestration-v2/WorktreeBranches.ts"; import * as TerminalManager from "./terminal/Manager.ts"; import { CodexProviderCapabilitiesV2 } from "./orchestration-v2/Adapters/CodexAdapterV2.ts"; import { ServerActivation } from "./serverActivation.ts"; @@ -349,10 +351,13 @@ const cleanupFixture = ( let outboxStatus: string | null = null; let fullStatusCalls = 0; let snapshotReads = 0; + const pullRequestLookups: string[] = []; let terminalListener: ((event: TerminalMetadataStreamEvent) => Effect.Effect) | undefined; let events: Stream.Stream = Stream.empty; const activated = yield* Deferred.make(); - const context = yield* Layer.build(StorageCleanup.layer).pipe( + const context = yield* Layer.build( + StorageCleanup.layer.pipe(Layer.provide(WorktreeBranches.layer)), + ).pipe( Effect.provideService(GitVcsDriver.GitVcsDriver, { ...git, execute: (input) => @@ -361,7 +366,8 @@ const cleanupFixture = ( // Windows the same status check the removal makes before deleting. if ( editBeforeRemoval && - (input.args.includes("remove") || input.args.includes("--porcelain")) + (input.args.includes("remove") || + (input.args.includes("--porcelain") && !input.args.includes("list"))) ) yield* fs.writeFileString(`${worktree}/tracked.txt`, "late edit\n"); return yield* git.execute(input); @@ -377,6 +383,14 @@ const cleanupFixture = ( } as unknown as Settings.ServerSettingsService["Service"]), Effect.provideService(ProjectStore.ProjectStoreV2, { list: () => Effect.sync(() => projectRoots.map((workspaceRoot) => ({ workspaceRoot }))), + getShell: (projectId: ProjectId) => + Effect.sync(() => { + const index = Number(projectId.replace("project-", "")) - 1; + const workspaceRoot = projectRoots[index]; + return workspaceRoot === undefined + ? Option.none() + : Option.some({ id: projectId, workspaceRoot }); + }), listShells: () => Effect.sync(() => projectRoots.map((workspaceRoot, index) => ({ @@ -420,6 +434,32 @@ const cleanupFixture = ( } as unknown as ProviderSessionManager.ProviderSessionManagerV2["Service"]), Effect.provideService(Orchestrator.OrchestratorV2, { streamDomainEvents: Stream.suspend(() => events), + getShellSnapshot: (input: { worktreePath: string }) => + Effect.sync(() => { + const members = threads.filter( + (thread) => thread.deletedAt === null && thread.worktreePath === input.worktreePath, + ); + return { + threads: members.filter((thread) => thread.archivedAt === null), + archivedThreads: members.filter((thread) => thread.archivedAt !== null), + }; + }), + dispatch: (command: { + threadId: ThreadId; + branch: string; + expectedBranch: string | null; + expectedWorktreePath: string; + }) => + Effect.gen(function* () { + const updatedAt = DateTime.makeUnsafe(yield* Clock.currentTimeMillis); + threads = threads.map((thread) => + thread.id === command.threadId && + thread.branch === command.expectedBranch && + thread.worktreePath === command.expectedWorktreePath + ? { ...thread, branch: command.branch, updatedAt } + : thread, + ); + }), } as unknown as Orchestrator.OrchestratorV2["Service"]), Effect.provideService(SqlClient.SqlClient, ((strings: TemplateStringsArray) => Effect.sync(() => @@ -456,6 +496,11 @@ const cleanupFixture = ( )) as unknown as SqlClient.SqlClient), Effect.provideService(GitManager.GitManager, { invalidateStatus: () => Effect.void, + branchPullRequest: (input: { branch: string }) => + Effect.sync(() => { + pullRequestLookups.push(input.branch); + return null; + }), } as unknown as GitManager.GitManager["Service"]), Effect.provideService(TerminalManager.TerminalManager, { subscribeMetadata: ( @@ -509,6 +554,7 @@ const cleanupFixture = ( }, }; }, + threads: () => threads, setThreads: (next: typeof threads) => { threads = next; }, @@ -551,6 +597,7 @@ const cleanupFixture = ( }) ?? Effect.void, fullStatusCalls: () => fullStatusCalls, snapshotReads: () => snapshotReads, + pullRequestLookups, activate: (domainEvents: Stream.Stream) => { events = domainEvents; return Deferred.succeed(activated, undefined); @@ -1064,7 +1111,7 @@ describe("subagent worktree ownership", () => { : shared ? "Shared by 2 threads" : scenario === "null-branch" || scenario === "stale-branch" - ? "A subagent thread's branch differs from this worktree's branch" + ? "Branch updated, re-checked next pass" : scenario === "recent-activity" ? "Settled for 2 of 8 days" : scenario === "recent-inactive" @@ -1073,6 +1120,14 @@ describe("subagent worktree ownership", () => { ? "Settled moments ago" : "Not settled", }); + if (scenario === "null-branch" || scenario === "stale-branch") { + expect(fixture.threads().map((thread) => thread.branch)).toEqual(["feature", "feature"]); + expect((yield* fixture.service.runNow).entries[0]).toMatchObject({ + outcome: "removed", + reason: "Settled for 20 days", + }); + yield* fixture.command(fixture.repo, ["rev-parse", "--verify", "refs/heads/feature"]); + } }).pipe(Effect.provide(cleanupTestLayer), Effect.scoped), ); @@ -1157,6 +1212,243 @@ describe("deleted thread worktrees", () => { ...overrides, }); + it.live.each(["detached", "default"] as const)( + "keeps a %s worktree until another checkout lets go of its recorded branch", + (kind) => + runCleanupTest( + Effect.gen(function* () { + const { service, fs, repo, worktree, command } = yield* cleanupFixture(); + if (kind === "default") yield* command(repo, ["switch", "--detach"]); + yield* command( + worktree, + kind === "default" ? ["switch", "main"] : ["switch", "--detach"], + ); + yield* command(repo, ["switch", "feature"]); + expect((yield* service.runNow).entries[0]).toMatchObject({ + outcome: "kept", + reason: "Another checkout holds the branch a thread here resumes on", + }); + yield* command(repo, ["switch", "--detach"]); + expect((yield* service.runNow).entries[0]).toMatchObject({ outcome: "removed" }); + expect(yield* fs.exists(worktree)).toBe(false); + yield* command(repo, ["worktree", "add", worktree, "feature"]); + }), + ), + ); + + it.live.each(["detached", "default", "switched-branch", "recorded-branch"] as const)( + "keeps a %s worktree while only its reflog holds a commit, until a ref holds it", + (kind) => + runCleanupTest( + Effect.gen(function* () { + const { service, fs, repo, worktree, command } = yield* cleanupFixture(); + if (kind === "detached" || kind === "default") + yield* command(worktree, ["switch", "--detach"]); + if (kind === "switched-branch") yield* command(worktree, ["switch", "-c", "agent/work"]); + yield* command(worktree, ["commit", "--allow-empty", "-m", "only the reflog holds this"]); + const orphan = (yield* command(worktree, ["rev-parse", "HEAD"])).stdout.trim(); + if (kind === "detached") yield* command(worktree, ["switch", "--detach", "main"]); + if (kind === "default") { + yield* command(repo, ["switch", "--detach"]); + yield* command(worktree, ["switch", "main"]); + } + if (kind === "switched-branch") { + yield* command(worktree, ["reset", "--hard", "HEAD~1"]); + expect((yield* service.runNow).entries[0]?.reason).toBe( + "Branch updated, re-checked next pass", + ); + } + if (kind === "recorded-branch") + yield* command(worktree, ["commit", "--amend", "--allow-empty", "-m", "rewritten"]); + expect((yield* service.runNow).entries[0]).toMatchObject({ + outcome: "kept", + reason: "Worktree history has commits no branch or tag holds", + }); + yield* command(repo, ["tag", "keeper", orphan]); + expect((yield* service.runNow).entries[0]).toMatchObject({ outcome: "removed" }); + expect(yield* fs.exists(worktree)).toBe(false); + yield* command(repo, ["cat-file", "-e", orphan]); + }), + ), + ); + + it.live("keeps a deleted owner's worktree while only its reflog holds a commit", () => + runCleanupTest( + Effect.gen(function* () { + const { service, fs, repo, worktree, command, setThreads, setDeletedThreads } = + yield* cleanupFixture({ + ...DEFAULT_SERVER_SETTINGS.storageCleanup, + worktreeAfterDays: null, + worktreeOnDelete: true, + }); + setThreads([]); + setDeletedThreads([parent(worktree)]); + yield* command(worktree, ["switch", "--detach"]); + yield* command(worktree, ["commit", "--allow-empty", "-m", "only the reflog holds this"]); + yield* command(repo, ["switch", "--detach"]); + yield* command(worktree, ["switch", "main"]); + expect((yield* service.runNow).entries[0]).toMatchObject({ + outcome: "kept", + reason: "Worktree history has commits no branch or tag holds", + }); + expect(yield* fs.exists(worktree)).toBe(true); + }), + ), + ); + + it.live.each(["live", "deleted"] as const)( + "keeps another repository's worktree at a %s owner's path, with its files", + (owner) => + runCleanupTest( + Effect.gen(function* () { + const { service, fs, repo, worktree, command, setThreads, setDeletedThreads } = + yield* cleanupFixture({ + ...DEFAULT_SERVER_SETTINGS.storageCleanup, + worktreeAfterDays: 8, + worktreeOnDelete: true, + worktreeKeepWhen: "tracked-changes", + }); + const other = yield* fs.realPath( + yield* fs.makeTempDirectoryScoped({ prefix: "cleanup-foreign-" }), + ); + yield* command(repo, ["worktree", "remove", "--force", worktree]); + yield* command(other, ["init", "-b", "main"]); + yield* command(other, ["config", "user.email", "foreign@example.test"]); + yield* command(other, ["config", "user.name", "Foreign"]); + yield* command(other, ["commit", "--allow-empty", "-m", "foreign"]); + yield* command(other, ["worktree", "add", "--detach", worktree]); + yield* fs.writeFileString(`${worktree}/precious.txt`, "mine\n"); + if (owner === "deleted") { + setThreads([]); + setDeletedThreads([parent(worktree)]); + } + expect((yield* service.runNow).entries[0]).toMatchObject({ + outcome: "kept", + reason: "Repository or branch changed", + }); + expect(yield* fs.readFileString(`${worktree}/precious.txt`)).toBe("mine\n"); + }), + ), + ); + + it.live("keeps detached worktrees while a thread in another one records the same branch", () => + runCleanupTest( + Effect.gen(function* () { + const { service, fs, repo, worktree, command, config, setThreads } = + yield* cleanupFixture(); + const second = `${config.worktreesDir}/feature-2`; + yield* command(repo, ["worktree", "add", "--detach", second]); + yield* command(worktree, ["switch", "--detach"]); + setThreads([ + shell({ branch: "feature", worktreePath: worktree }), + shell({ + id: ThreadId.make("thread-2"), + branch: "feature", + worktreePath: second, + lineage: { + rootThreadId: ThreadId.make("thread-2"), + parentThreadId: null, + relationshipToParent: null, + }, + }), + ]); + const report = yield* service.runNow; + expect(report.entries.map((entry) => entry.outcome)).toEqual(["kept", "kept"]); + expect(report.entries[0]?.reason).toBe( + "A thread in another worktree records the same branch", + ); + expect(yield* fs.exists(worktree)).toBe(true); + expect(yield* fs.exists(second)).toBe(true); + }), + ), + ); + + it.live("rebinds a thread that stores its worktree path with a trailing slash", () => + runCleanupTest( + Effect.gen(function* () { + const { service, fs, worktree, command, threads, setThreads } = yield* cleanupFixture(); + setThreads([shell({ branch: "feature", worktreePath: `${worktree}/` })]); + yield* command(worktree, ["switch", "-c", "agent/work"]); + expect((yield* service.runNow).entries[0]?.reason).toBe( + "Branch updated, re-checked next pass", + ); + expect(threads()[0]?.branch).toBe("agent/work"); + expect((yield* service.runNow).entries[0]).toMatchObject({ outcome: "removed" }); + expect(yield* fs.exists(worktree)).toBe(false); + }), + ), + ); + + it.live("rebinds an owner with no recorded branch, and keeps one it cannot rebind", () => + runCleanupTest( + Effect.gen(function* () { + const { service, fs, worktree, command, threads, setThreads } = yield* cleanupFixture(); + setThreads([shell({ branch: null, worktreePath: worktree })]); + yield* command(worktree, ["switch", "--detach"]); + expect((yield* service.runNow).entries[0]).toMatchObject({ + outcome: "kept", + reason: "Thread has no branch to resume on", + }); + yield* command(worktree, ["switch", "-c", "agent/work"]); + expect((yield* service.runNow).entries[0]?.reason).toBe( + "Branch updated, re-checked next pass", + ); + expect(threads()[0]?.branch).toBe("agent/work"); + expect((yield* service.runNow).entries[0]).toMatchObject({ outcome: "removed" }); + expect(yield* fs.exists(worktree)).toBe(false); + }), + ), + ); + + it.live.each([ + ["detached", null], + ["default", null], + ["default", "main"], + ] as const)( + "applies the usual rules to a %s worktree whose subagent records %s", + ([kind, subagentBranch]) => + runCleanupTest( + Effect.gen(function* () { + const { service, fs, repo, worktree, command, setThreads } = yield* cleanupFixture(); + const owner = shell({ branch: "feature", worktreePath: worktree }); + setThreads([owner, subagent(owner, { branch: subagentBranch })]); + if (kind === "default") yield* command(repo, ["switch", "--detach"]); + yield* command( + worktree, + kind === "default" ? ["switch", "main"] : ["switch", "--detach"], + ); + expect((yield* service.runNow).entries[0]).toMatchObject({ outcome: "removed" }); + expect(yield* fs.exists(worktree)).toBe(false); + }), + ), + ); + + it.live("removes a deleted owner's worktree after rebinding its surviving subagent", () => + runCleanupTest( + Effect.gen(function* () { + const fixture = yield* cleanupFixture({ + ...DEFAULT_SERVER_SETTINGS.storageCleanup, + worktreeAfterDays: null, + worktreeOnDelete: true, + }); + const owner = parent(fixture.worktree); + fixture.setDeletedThreads([owner]); + fixture.setThreads([subagent(owner)]); + yield* fixture.command(fixture.worktree, ["switch", "-c", "agent/work"]); + expect((yield* fixture.service.runNow).entries[0]).toMatchObject({ + outcome: "kept", + reason: "Branch updated, re-checked next pass", + }); + expect(fixture.threads()[0]?.branch).toBe("agent/work"); + expect((yield* fixture.service.runNow).entries[0]).toMatchObject({ + outcome: "removed", + reason: "Thread was deleted", + }); + yield* fixture.command(fixture.repo, ["rev-parse", "--verify", "refs/heads/agent/work"]); + }), + ), + ); + it.effect.each(["no subagents", "native subagent"] as const)( "removes a deleted thread's worktree with %s", (scenario) => @@ -2181,7 +2473,14 @@ describe("storage cleanup reports and local file policies", () => { yield* cleanupFixture(); const changed = { outcome: "kept", reason: "Repository or branch changed" }; yield* command(worktree, ["checkout", "--detach"]); - expect((yield* service.runNow).entries[0]).toMatchObject(changed); + yield* command(worktree, ["commit", "--allow-empty", "-m", "detached work"]); + const detached = (yield* command(worktree, ["rev-parse", "HEAD"])).stdout.trim(); + expect((yield* service.runNow).entries[0]).toMatchObject({ + outcome: "kept", + reason: "Worktree history has commits no branch or tag holds", + }); + yield* command(repo, ["cat-file", "-e", detached]); + yield* command(repo, ["tag", "keeper", detached]); yield* command(worktree, ["switch", "--orphan", "other"]); expect((yield* service.runNow).entries[0]).toMatchObject(changed); yield* command(worktree, ["switch", "-c", "(feature)", "feature"]); @@ -2214,6 +2513,192 @@ describe("storage cleanup reports and local file policies", () => { }), ), ); + it.live("removes a detached worktree only once a ref outside refs/t3 holds its commit", () => + runCleanupTest( + Effect.gen(function* () { + const { service, fs, repo, worktree, command } = yield* cleanupFixture(); + yield* command(worktree, ["switch", "--detach"]); + yield* command(worktree, ["commit", "--allow-empty", "-m", "detached work"]); + const detached = (yield* command(worktree, ["rev-parse", "HEAD"])).stdout.trim(); + yield* command(repo, ["update-ref", "refs/t3/checkpoints/run", detached]); + expect((yield* service.runNow).entries[0]).toMatchObject({ + outcome: "kept", + reason: "Worktree history has commits no branch or tag holds", + }); + expect(yield* fs.exists(worktree)).toBe(true); + yield* command(repo, ["tag", "keeper", detached]); + expect((yield* service.runNow).entries[0]).toMatchObject({ outcome: "removed" }); + yield* command(repo, ["cat-file", "-e", detached]); + }), + ), + ); + it.live.each([ + ["a remote branch a prune can drop", "agent"], + ["the remote default branch, which a stale local HEAD can misname", "main"], + ] as const)("keeps a detached commit held only by %s", ([, remoteBranch]) => + runCleanupTest( + Effect.gen(function* () { + const { service, fs, repo, worktree, command } = yield* cleanupFixture(); + const remote = yield* fs.makeTempDirectoryScoped({ prefix: "cleanup-remote-" }); + yield* command(remote, ["init", "--bare", "-b", "main"]); + yield* command(repo, ["remote", "add", "origin", remote]); + yield* command(repo, ["push", "origin", "main"]); + yield* command(repo, ["remote", "set-head", "origin", "main"]); + yield* command(worktree, ["switch", "--detach"]); + yield* command(worktree, ["commit", "--allow-empty", "-m", "detached work"]); + yield* command(worktree, ["push", "origin", `HEAD:refs/heads/${remoteBranch}`]); + expect((yield* service.runNow).entries[0]).toMatchObject({ + outcome: "kept", + reason: "Worktree history has commits no branch or tag holds", + }); + expect(yield* fs.exists(worktree)).toBe(true); + }), + ), + ); + it.live("writes no branch for a thread outside the managed worktree folder", () => + runCleanupTest( + Effect.gen(function* () { + const { service, fs, repo, command, threads, setThreads } = yield* cleanupFixture(); + const outside = yield* fs.realPath( + yield* fs.makeTempDirectoryScoped({ prefix: "cleanup-outside-" }), + ); + yield* command(repo, ["worktree", "add", "-b", "scratch-work", `${outside}/scratch`]); + setThreads([shell({ branch: null, worktreePath: `${outside}/scratch` })]); + expect((yield* service.runNow).entries[0]).toMatchObject({ + outcome: "kept", + reason: "Outside the managed worktree folder", + }); + expect(threads()[0]?.branch).toBeNull(); + }), + ), + ); + it.live("follows a branch the agent switched to and removes the worktree next pass", () => + runCleanupTest( + Effect.gen(function* () { + const { service, fs, repo, worktree, command, threads } = yield* cleanupFixture(); + yield* command(worktree, ["switch", "-c", "agent/work"]); + yield* command(worktree, ["commit", "--allow-empty", "-m", "agent work"]); + const tip = (yield* command(worktree, ["rev-parse", "HEAD"])).stdout.trim(); + expect((yield* service.runNow).entries[0]).toMatchObject({ + outcome: "kept", + reason: "Branch updated, re-checked next pass", + }); + expect(threads()[0]?.branch).toBe("agent/work"); + expect(yield* fs.exists(worktree)).toBe(true); + expect((yield* service.runNow).entries[0]).toMatchObject({ outcome: "removed" }); + expect(yield* fs.exists(worktree)).toBe(false); + expect((yield* command(repo, ["rev-parse", "refs/heads/agent/work"])).stdout.trim()).toBe( + tip, + ); + }), + ), + ); + it.live("removes a worktree on the default branch without rebinding its thread", () => + runCleanupTest( + Effect.gen(function* () { + const { service, fs, repo, worktree, command, threads } = yield* cleanupFixture(); + yield* command(repo, ["switch", "--detach"]); + yield* command(worktree, ["switch", "main"]); + expect((yield* service.runNow).entries[0]).toMatchObject({ outcome: "removed" }); + expect(yield* fs.exists(worktree)).toBe(false); + expect(threads()[0]?.branch).toBe("feature"); + }), + ), + ); + it.live.each([ + ["the branch the threads now record", "agent", ["agent/work"]], + ["no branch for a detached commit", "detached", []], + ["no branch for the default branch", "default", []], + ] as const)("looks up a pull request for %s", ([, checkout, lookups]) => + runCleanupTest( + Effect.gen(function* () { + const { service, fs, repo, worktree, command, pullRequestLookups } = yield* cleanupFixture({ + ...DEFAULT_SERVER_SETTINGS.storageCleanup, + worktreeOnMerge: true, + }); + const remote = yield* fs.makeTempDirectoryScoped({ prefix: "cleanup-remote-" }); + yield* command(remote, ["init", "--bare", "-b", "main"]); + yield* command(repo, ["remote", "add", "origin", remote]); + yield* command(repo, ["push", "origin", "main"]); + yield* command(repo, ["remote", "set-head", "origin", "main"]); + if (checkout === "agent") yield* command(worktree, ["switch", "-c", "agent/work"]); + if (checkout === "detached") yield* command(worktree, ["switch", "--detach"]); + if (checkout === "default") { + yield* command(repo, ["switch", "--detach"]); + yield* command(worktree, ["switch", "main"]); + } + yield* service.runNow; + expect((yield* service.runNow).entries[0]).toMatchObject({ + outcome: "kept", + reason: "Not merged", + }); + expect(pullRequestLookups).toEqual(lookups); + }), + ), + ); + it.live.each([ + ["the commit moves", "Commit changed since check"], + ["the branch switches", "Repository or branch changed since check"], + [ + "the detached commit loses its holding ref", + "Worktree history has commits no branch or tag holds", + ], + ] as const)("keeps a worktree when %s during size measurement", ([change, reason]) => + runCleanupTest( + Effect.gen(function* () { + const fixture = yield* pausedMeasurementFixture(); + const { command, repo, worktree } = fixture; + const heldOnlyByTag = change === "the detached commit loses its holding ref"; + if (heldOnlyByTag) { + yield* command(worktree, ["switch", "--detach"]); + yield* command(worktree, ["commit", "--allow-empty", "-m", "detached work"]); + yield* command(worktree, ["tag", "keeper"]); + } + const run = yield* fixture.service.runNow.pipe(Effect.forkChild); + yield* Deferred.await(fixture.measuring); + if (heldOnlyByTag) yield* command(repo, ["tag", "--delete", "keeper"]); + if (change === "the commit moves") + yield* command(worktree, ["commit", "--allow-empty", "-m", "late work"]); + if (change === "the branch switches") yield* command(worktree, ["switch", "-c", "late"]); + yield* Deferred.succeed(fixture.resume, undefined); + expect((yield* Fiber.join(run)).entries[0]).toMatchObject({ outcome: "kept", reason }); + expect(yield* fixture.fs.exists(worktree)).toBe(true); + }), + ), + ); + it.live("keeps a detached worktree when its holding ref goes during the final checks", () => + runCleanupTest( + Effect.gen(function* () { + const git = yield* GitVcsDriver.GitVcsDriver; + let localChangeChecks = 0; + let repo = ""; + const { service, command, worktree, ...fixture } = yield* cleanupFixture().pipe( + Effect.provideService(GitVcsDriver.GitVcsDriver, { + ...git, + execute: (input) => + Effect.gen(function* () { + if (input.operation === "StorageCleanup.localChanges" && ++localChangeChecks === 2) + yield* git.execute({ + operation: "test", + cwd: repo, + args: ["tag", "--delete", "keeper"], + }); + return yield* git.execute(input); + }), + }), + ); + repo = fixture.repo; + yield* command(worktree, ["switch", "--detach"]); + yield* command(worktree, ["commit", "--allow-empty", "-m", "detached work"]); + yield* command(worktree, ["tag", "keeper"]); + expect((yield* service.runNow).entries[0]).toMatchObject({ + outcome: "kept", + reason: "Worktree history has commits no branch or tag holds", + }); + expect(yield* fixture.fs.exists(worktree)).toBe(true); + }), + ), + ); it.live.each([ ["any-local-files", ".env", "kept"], ["uncommitted-changes", ".env", "removed"], diff --git a/apps/server/src/storageCleanup.ts b/apps/server/src/storageCleanup.ts index f80b02a371eae..07ddde641a4cc 100644 --- a/apps/server/src/storageCleanup.ts +++ b/apps/server/src/storageCleanup.ts @@ -42,6 +42,7 @@ import * as Orchestrator from "./orchestration-v2/Orchestrator.ts"; import * as ProjectionStore from "./orchestration-v2/ProjectionStore.ts"; import * as ProviderSessionManager from "./orchestration-v2/ProviderSessionManager.ts"; import { threadHasQueuedTurnStart } from "./orchestration-v2/ThreadSettlementService.ts"; +import * as WorktreeBranches from "./orchestration-v2/WorktreeBranches.ts"; import { forkParked } from "./serverActivation.ts"; import * as Settings from "./serverSettings.ts"; import * as TerminalManager from "./terminal/Manager.ts"; @@ -301,6 +302,7 @@ const make = Effect.gen(function* () { const providerSessions = yield* ProviderSessionManager.ProviderSessionManagerV2; const sql = yield* SqlClient.SqlClient; const git = yield* GitVcsDriver.GitVcsDriver; + const worktreeBranches = yield* WorktreeBranches.WorktreeBranches; const gitManager = yield* GitManager.GitManager; const terminals = yield* TerminalManager.TerminalManager; const fs = yield* FileSystem.FileSystem; @@ -566,62 +568,6 @@ const make = Effect.gen(function* () { } }); - // Use two small Git reads: statusDetailsLocal also diffs and counts for every worktree. - const branchChanged = Effect.fn("StorageCleanup.branchChanged")(function* ( - cwd: string, - branch: string | null, - ) { - const failure = (detail: string) => - new GitCommandError({ - operation: "StorageCleanup.branchChanged", - command: "git", - cwd, - detail, - }); - // Return null for missing/non-repository cwd; force untranslated git diagnostics for matching. - const read = Effect.fnUntraced(function* (args: string[], allowedExitCode?: number) { - const result = yield* git - .execute({ - operation: "StorageCleanup.branchChanged", - cwd, - args, - env: { LC_ALL: "C" }, - allowNonZeroExit: true, - }) - .pipe( - Effect.catch((error) => - Effect.flatMap(fs.exists(cwd), (exists) => - exists ? Effect.fail(error) : Effect.succeed(null), - ), - ), - ); - if (result === null) return null; - if (result.exitCode === 0 || result.exitCode === allowedExitCode) return result; - const stderr = result.stderr.toLowerCase(); - // Newer Git reports a dangling .git file as "gitfile does not point to a valid repository". - if ( - stderr.includes("not a git repository") || - stderr.includes("does not point to a valid repository") - ) - return null; - return yield* failure( - result.stderr.trim().split(/\r?\n/)[0]?.slice(0, 1000) || "command failed", - ); - }); - const index = yield* read(["rev-parse", "--git-path", "index"]); - if (index === null) return true; - if (yield* fs.exists(`${path.resolve(cwd, index.stdout.trim())}.lock`)) - return yield* failure( - "Git index is locked. Status will resume when the index lock is removed.", - ); - // Detached HEAD exits 1; full refs avoid shortening when a tag shares the branch name. - const head = yield* read(["symbolic-ref", "--quiet", "HEAD"], 1); - if (head === null) return true; - const ref = head.exitCode === 0 ? head.stdout.trim() : ""; - const name = ref.startsWith("refs/heads/") ? ref.slice("refs/heads/".length).trim() : null; - return (name?.startsWith("(") ? null : name) !== branch; - }); - const localChangesReason = Effect.fn("StorageCleanup.localChangesReason")(function* ( cwd: string, rules: WorktreeCleanupRules, @@ -662,6 +608,43 @@ const make = Effect.gen(function* () { return file === undefined ? null : `has ignored files (${file})`; }); + // Removing a worktree loses its reflog, the last path to commits no branch holds. Rewritten + // commits land there too, but gc expires them, so that only delays removal. A thread can resume + // only if `git worktree add` can still check out its recorded branch. + const checkoutBlocker = Effect.fn("StorageCleanup.checkoutBlocker")(function* ( + cwd: string, + head: string, + checkedOut: string | null, + recorded: ReadonlyArray, + ) { + const reflog = yield* git.execute({ + operation: "StorageCleanup.checkoutBlocker", + cwd, + args: ["reflog", "show", "--format=%H", "HEAD"], + maxOutputBytes: 16 * 1024 * 1024, + }); + // Only refs that stay count. A user cannot recover from refs/t3, and a routine + // `git fetch --prune` can drop any remote branch, even one a stale local HEAD calls the default. + const unheld = yield* git.execute({ + operation: "StorageCleanup.checkoutBlocker", + cwd, + args: ["rev-list", "-n", "1", "--ignore-missing", "--stdin", "--not", "--branches", "--tags"], + stdin: `${head}\n${reflog.stdout}`, + }); + if (unheld.stdout.trim() !== "") return "worktree history has commits no branch or tag holds"; + const worktrees = yield* git.execute({ + operation: "StorageCleanup.checkoutBlocker", + cwd, + args: ["worktree", "list", "--porcelain"], + }); + const held = new Set( + [...worktrees.stdout.matchAll(/^branch refs\/heads\/(.+?)\r?$/gm)].map((match) => match[1]), + ); + return recorded.some((name) => name !== null && name !== checkedOut && held.has(name)) + ? "another checkout holds the branch a thread here resumes on" + : null; + }); + const submoduleRemoval = Effect.fn("StorageCleanup.submoduleRemoval")( function* (cwd: string, rules: WorktreeCleanupRules) { const inspect = (args: string[], maxOutputBytes = 64 * 1024) => @@ -901,13 +884,8 @@ const make = Effect.gen(function* () { if (members.length > 1 && worktreeOwner(members) === undefined) return keep(`shared by ${members.length} threads`); if (project === undefined) return keep("project is unavailable"); - if ( - thread.branch === null || - !group.every((member) => storageCleanupThreadIdle(member, now, serverSettings)) - ) + if (!group.every((member) => storageCleanupThreadIdle(member, now, serverSettings))) return keep(waitForStop("thread is running or has pending work")); - if (group.some((member) => member.branch !== thread.branch)) - return keep("a subagent thread's branch differs from this worktree's branch"); if (hasTerminal(worktreePath)) return keep(waitForStop("open terminal")); // Compare canonical roots while allowing symlinked parents, never symlinked worktree directories. const realPath = yield* fs.realPath(worktreePath); @@ -920,8 +898,17 @@ const make = Effect.gen(function* () { return keep("contains a project checkout"); if ((yield* fs.stat(path.join(worktreePath, ".git"))).type !== "File") return keep("not a linked worktree"); - if (yield* branchChanged(worktreePath, thread.branch)) + // Before anything reads the checkout: another repository's worktree here is not ours to judge. + if (!(yield* worktreeBranches.sameRepository(worktreePath, project.workspaceRoot))) return keep("repository or branch changed"); + const checkout = yield* worktreeBranches.sync({ + worktreePath, + // `groups` matches resolved paths, but a thread can store another spelling of one. + threadPaths: [...new Set(group.map((member) => member.worktreePath!))], + }); + if (checkout._tag === "Unavailable") return keep("repository or branch changed"); + if (checkout._tag === "Rebinding") return keep("branch updated, re-checked next pass"); + if (thread.branch === null) return keep("thread has no branch to resume on"); const changes = yield* localChangesReason(worktreePath, settings); if (changes !== null) return keep(changes); if ( @@ -934,6 +921,26 @@ const make = Effect.gen(function* () { ) return keep(waitForStop("settled moments ago")); const head = yield* git.resolveCommit({ cwd: worktreePath, revision: "HEAD" }); + const checkCheckout = () => + checkoutBlocker( + worktreePath, + head.commitSha, + checkout._tag === "Detached" ? null : checkout.branch, + group.map((member) => member.branch), + ); + const blocker = yield* checkCheckout(); + if (blocker !== null) return keep(blocker); + // Only one checkout can hold a branch, so a second thread could not resume on it. + if ( + checkout._tag !== "Bound" && + snapshot.threads.some( + (other) => + other.worktreePath !== null && + path.resolve(other.worktreePath) !== worktreePath && + group.some((member) => member.branch !== null && member.branch === other.branch), + ) + ) + return keep("a thread in another worktree records the same branch"); const old = !deleted && settings.worktreeAfterDays !== null && @@ -994,10 +1001,7 @@ const make = Effect.gen(function* () { timeoutMs: 20_000, }) .pipe( - Effect.map( - (result) => - /^ref: refs\/heads\/(.+)\tHEAD\r?$/m.exec(result.stdout)?.[1] ?? null, - ), + Effect.map((result) => GitVcsDriver.parseRemoteHeadBranch(result.stdout)), Effect.orElseSucceed(() => null), ); } @@ -1033,13 +1037,18 @@ const make = Effect.gen(function* () { remote === null ? "no commits beyond the project branch" : "no commits beyond the default branch"; - if (!eligible && remote !== null && settings.worktreeOnMerge && thread.branch !== null) { + if ( + !eligible && + remote !== null && + settings.worktreeOnMerge && + checkout._tag === "Bound" + ) { const pullRequest = yield* gitManager.branchPullRequest( - { cwd: worktreePath, branch: thread.branch }, + { cwd: worktreePath, branch: checkout.branch }, { refresh: true }, ); eligible = storageCleanupPullRequestMerged(pullRequest, { - branch: thread.branch, + branch: checkout.branch, defaultBranch: branch, headSha: head.commitSha, integrated, @@ -1093,12 +1102,11 @@ const make = Effect.gen(function* () { if ( latestGroup.length !== group.length || latestOwner?.id !== thread.id || - latestOwner?.branch !== thread.branch || latestGroup.some((shell) => { const previous = group.find((member) => member.id === shell.id); return ( previous === undefined || - shell.branch !== thread.branch || + shell.branch !== previous.branch || !Equal.equals(shell.settledAt, previous.settledAt) || (shell.settledAt !== null && !Equal.equals(shell.updatedAt, previous.updatedAt)) || !storageCleanupThreadIdle(shell, now, serverSettings) || @@ -1153,7 +1161,12 @@ const make = Effect.gen(function* () { const bytes = yield* measureWorktree(worktreePath); const blockedWhileMeasuring = yield* blockedSinceCheck(); if (blockedWhileMeasuring !== null) return keep(blockedWhileMeasuring); - if (yield* branchChanged(worktreePath, thread.branch)) + const observed = yield* worktreeBranches.observe(worktreePath); + if ( + checkout._tag === "Detached" + ? observed._tag !== "Detached" + : observed._tag !== "Named" || observed.branch !== checkout.branch + ) return keep("repository or branch changed since check"); if ( (yield* git.resolveCommit({ cwd: worktreePath, revision: "HEAD" })).commitSha !== @@ -1182,6 +1195,9 @@ const make = Effect.gen(function* () { const changes = yield* localChangesReason(worktreePath, settings); if (changes !== null) return keep(changes); } + // Last, so nothing that runs after this can drop the ref that holds the commit. + const finalBlocker = yield* checkCheckout(); + if (finalBlocker !== null) return keep(finalBlocker); yield* removeLinkedWorktree({ operation: "StorageCleanup.cleanupGit", cwd: project.workspaceRoot, diff --git a/apps/server/src/vcs/GitVcsDriver.ts b/apps/server/src/vcs/GitVcsDriver.ts index a2cd548bcc0b9..a23f6576ebbab 100644 --- a/apps/server/src/vcs/GitVcsDriver.ts +++ b/apps/server/src/vcs/GitVcsDriver.ts @@ -301,6 +301,15 @@ export interface GitRemoteStatusOptions { readonly refreshUpstream?: boolean; } +export const isDefaultBranchName = (branch: string | null, defaultBranch: string | null) => + branch !== null && + (branch === defaultBranch || + (defaultBranch === null && (branch === "main" || branch === "master"))); + +/** The default branch in the output of `git ls-remote --symref HEAD`. */ +export const parseRemoteHeadBranch = (stdout: string) => + /^ref: refs\/heads\/(.+)\tHEAD\r?$/m.exec(stdout)?.[1] ?? null; + export class GitVcsDriver extends Context.Service< GitVcsDriver, { diff --git a/apps/server/src/vcs/GitVcsDriverCore.ts b/apps/server/src/vcs/GitVcsDriverCore.ts index 94404b488b136..776ccf7dd1f19 100644 --- a/apps/server/src/vcs/GitVcsDriverCore.ts +++ b/apps/server/src/vcs/GitVcsDriverCore.ts @@ -2068,10 +2068,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* } const defaultBranch = yield* resolveDefaultBranchName(cwd, "origin"); - const isDefaultBranch = - branch !== null && - (branch === defaultBranch || - (defaultBranch === null && (branch === "main" || branch === "master"))); + const isDefaultBranch = GitVcsDriver.isDefaultBranchName(branch, defaultBranch); const aheadOfDefaultCount = branch && !isDefaultBranch ? upstreamRef === null @@ -2294,10 +2291,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* } } - const isDefaultBranch = - refName !== null && - (refName === defaultBranch || - (defaultBranch === null && (refName === "main" || refName === "master"))); + const isDefaultBranch = GitVcsDriver.isDefaultBranchName(refName, defaultBranch); const needsFallbackAhead = includeDivergence && !upstreamRef && refName !== null; const needsAheadOfDefault = includeDivergence && refName !== null && !isDefaultBranch; const [baseAheadCount, branchChanges] = yield* Effect.all(