From 8f36cd8707042aca510be3b5685c558ddf0acdbf Mon Sep 17 00:00:00 2001 From: Luiz Ferraz Date: Sat, 12 Sep 2026 19:05:12 +0000 Subject: [PATCH] feat(server): remove worktrees when their last thread settles - removes a settled thread's worktree once no active threads share it and its session stops - preserves the project root, branches, and dirty worktrees - recreates removed worktrees from the saved branch on the next turn --- .../ThreadSettlementReactor.test.ts | 279 ++++++++++++++++++ .../orchestration/ThreadSettlementReactor.ts | 80 ++++- apps/server/src/vcs/GitVcsDriverCore.test.ts | 1 + docs/user/thread-sidebar.md | 4 + 4 files changed, 362 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration/ThreadSettlementReactor.test.ts b/apps/server/src/orchestration/ThreadSettlementReactor.test.ts index 0690eea2d50e..e588f8cf9e68 100644 --- a/apps/server/src/orchestration/ThreadSettlementReactor.test.ts +++ b/apps/server/src/orchestration/ThreadSettlementReactor.test.ts @@ -1,10 +1,13 @@ import { DEFAULT_SERVER_SETTINGS, + EventId, + GitCommandError, ProjectId, ProviderInstanceId, PullRequestOperationError, ThreadId, type OrchestrationCommand, + type OrchestrationEvent, type OrchestrationProjectShell, type OrchestrationShellSnapshot, type OrchestrationThreadShell, @@ -19,6 +22,7 @@ import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; import * as PubSub from "effect/PubSub"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; @@ -26,6 +30,7 @@ import * as Stream from "effect/Stream"; import { TestClock } from "effect/testing"; import { GitManager, type GitBranchPullRequest } from "../git/GitManager.ts"; +import { GitWorkflowService } from "../git/GitWorkflowService.ts"; import { PullRequestService, type PullRequestMergeEvent, @@ -110,6 +115,22 @@ function makeSnapshot( }; } +function settledEvent(threadId: ThreadId): OrchestrationEvent { + return { + type: "thread.settled", + sequence: 2, + eventId: EventId.make(`settled:${threadId}`), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: NOW, + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + payload: { threadId, settledAt: NOW, updatedAt: NOW }, + }; +} + function makePullRequestSummary(input: { readonly projectId: ProjectId; readonly repository: string; @@ -157,6 +178,9 @@ interface HarnessOptions { readonly branchPullRequest?: GitManager["Service"]["branchPullRequest"]; readonly pullRequestSummary?: PullRequestService["Service"]["summary"]; readonly existingWorktreePaths?: ReadonlyArray; + readonly canonicalPaths?: Readonly>; + readonly removeWorktree?: GitWorkflowService["Service"]["removeWorktree"]; + readonly publishSettlements?: boolean; readonly onDispatch?: ( command: AutoSettleCommand, ) => Effect.Effect; @@ -171,6 +195,10 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: const settingsReads = yield* Queue.unbounded(); const settingsChanges = yield* PubSub.unbounded(); const mergedPullRequests = yield* PubSub.unbounded(); + const domainEvents = yield* PubSub.unbounded(); + const removedWorktrees = yield* Ref.make< + Array[0]> + >([]); const commands = yield* Ref.make>([]); const branchCalls = yield* Ref.make< ReadonlyArray<{ readonly cwd: string; readonly branch: string }> @@ -221,10 +249,22 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: } return Ref.update(commands, (recorded) => [...recorded, command]).pipe( Effect.andThen(options.onDispatch?.(command) ?? Effect.void), + Effect.tap(() => (options.publishSettlements ? settle(command.threadId) : Effect.void)), Effect.as({ sequence: 1 }), ); }; + const publishEvent = (event: OrchestrationEvent) => PubSub.publish(domainEvents, event); + const settle = (threadId: ThreadId) => + Ref.update(snapshots, (snapshot) => ({ + ...snapshot, + threads: snapshot.threads.map((thread) => + thread.id === threadId + ? { ...thread, settledOverride: "settled" as const, settledAt: NOW } + : thread, + ), + })).pipe(Effect.andThen(publishEvent(settledEvent(threadId)))); + const serverSettings = ServerSettingsService.of({ start: Effect.void, ready: Effect.void, @@ -248,6 +288,12 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: branchPullRequest, invalidateStatus: (cwd) => Ref.update(invalidatedCwds, (cwds) => [...cwds, cwd]), }), + Layer.mock(GitWorkflowService)({ + removeWorktree: (input) => + Ref.update(removedWorktrees, (calls) => [...calls, input]).pipe( + Effect.andThen(options.removeWorktree?.(input) ?? Effect.void), + ), + }), Layer.mock(PullRequestService)({ summary: pullRequestSummary, subscribeMerges: PubSub.subscribe(mergedPullRequests).pipe( @@ -258,6 +304,9 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: readEvents: () => Stream.empty, dispatch, streamDomainEvents: Stream.empty, + subscribeDomainEvents: PubSub.subscribe(domainEvents).pipe( + Effect.map((subscription) => Stream.fromSubscription(subscription)), + ), latestSequence: Effect.succeed(0), }), Layer.succeed(ServerSettingsService, serverSettings), @@ -265,7 +314,9 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: Layer.succeed(Crypto.Crypto, testCrypto), FileSystem.layerNoop({ exists: (path) => Effect.succeed(options.existingWorktreePaths?.includes(path) ?? false), + realPath: (path) => Effect.succeed(options.canonicalPaths?.[path] ?? path), }), + Path.layer, ); return { @@ -279,6 +330,9 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: summaryCalls, summaryRecovery, invalidatedCwds, + removedWorktrees, + publishEvent, + settle, updateSettings, publishMerge: PubSub.publish(mergedPullRequests, { projectId: PROJECT_ID, @@ -302,6 +356,231 @@ const startHarness = Effect.fn("startThreadSettlementHarness")(function* ( }); describe("ThreadSettlementReactor", () => { + const worktreePath = "/workspace/feature"; + const cleanupSettings = { + ...DEFAULT_SERVER_SETTINGS, + sidebarAutoSettleAfterDays: null, + sidebarAutoSettleOnMerge: false, + }; + + it.effect.each([ + { name: "main worktree", path: null, others: [], removed: false }, + { name: "explicit project root", path: "/workspace/project/.", others: [], removed: false }, + { name: "project root alias", path: "/workspace/root-alias", others: [], removed: false }, + { name: "unused worktree", path: worktreePath, others: [], removed: true }, + { + name: "another active thread", + path: worktreePath, + others: [makeThread("other", { worktreePath })], + removed: false, + }, + { + name: "active thread through an alias", + path: worktreePath, + others: [makeThread("other", { worktreePath: "/workspace/feature-alias" })], + removed: false, + }, + { + name: "active thread in another project root", + path: worktreePath, + others: [makeThread("other", { projectId: LINKED_PROJECT_ID })], + removed: false, + }, + { + name: "another settled thread", + path: worktreePath, + others: [makeThread("other", { worktreePath, settledOverride: "settled", settledAt: NOW })], + removed: true, + }, + { + name: "archived thread", + path: worktreePath, + others: [makeThread("other", { worktreePath, archivedAt: NOW })], + removed: true, + }, + { + name: "active thread in a different worktree", + path: worktreePath, + others: [makeThread("other", { worktreePath: "/workspace/different" })], + removed: true, + }, + ])("cleans up manual settlement with $name", ({ path, others, removed }) => + Effect.scoped( + Effect.gen(function* () { + const thread = makeThread("manual", { worktreePath: path, branch: "feature" }); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot( + [thread, ...others], + [makeProject(), makeProject(LINKED_PROJECT_ID, worktreePath)], + ), + settings: cleanupSettings, + canonicalPaths: { + "/workspace/root-alias": "/workspace/project", + "/workspace/feature-alias": worktreePath, + }, + }); + yield* Effect.gen(function* () { + const reactor = yield* ThreadSettlementReactor.ThreadSettlementReactor; + yield* reactor.start(); + yield* Deferred.succeed(fixture.activation, undefined); + yield* fixture.settle(thread.id); + yield* Queue.take(fixture.snapshotReads); + yield* reactor.drain; + assert.deepStrictEqual( + yield* Ref.get(fixture.removedWorktrees), + removed ? [{ cwd: "/workspace/project", path: worktreePath }] : [], + ); + const current = (yield* Ref.get(fixture.snapshots)).threads[0]!; + assert.strictEqual(current.branch, "feature"); + assert.strictEqual(current.worktreePath, path); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + + it.effect("cleans up after a merged pull request automatically settles a thread", () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const removed = yield* Deferred.make(); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([ + makeThread("merged", { + worktreePath, + branch: "feature", + latestUserMessageAt: "2026-08-27T00:00:00.000Z", + }), + ]), + settings: { ...cleanupSettings, sidebarAutoSettleOnMerge: true }, + branchPullRequest: () => Effect.succeed(makeBranchPullRequest("merged")), + publishSettlements: true, + removeWorktree: () => Deferred.succeed(removed, undefined).pipe(Effect.asVoid), + }); + yield* Effect.gen(function* () { + const reactor = yield* ThreadSettlementReactor.ThreadSettlementReactor; + yield* startHarness(reactor, fixture.activation, fixture.snapshotReads); + yield* Deferred.await(removed); + yield* reactor.drain; + assert.deepStrictEqual(yield* Ref.get(fixture.removedWorktrees), [ + { cwd: "/workspace/project", path: worktreePath }, + ]); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + + it.effect("waits for the last thread sharing a worktree to settle and stop its session", () => + Effect.scoped( + Effect.gen(function* () { + const first = makeThread("first", { worktreePath }); + const second = makeThread("second", { + worktreePath, + session: { + threadId: ThreadId.make("second"), + status: "ready", + providerName: "codex", + runtimeMode: "full-access", + activeTurnId: null, + lastError: null, + updatedAt: NOW, + }, + }); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([first, second]), + settings: cleanupSettings, + }); + yield* Effect.gen(function* () { + const reactor = yield* ThreadSettlementReactor.ThreadSettlementReactor; + yield* reactor.start(); + yield* Deferred.succeed(fixture.activation, undefined); + for (const thread of [first, second]) { + yield* fixture.settle(thread.id); + yield* Queue.take(fixture.snapshotReads); + yield* reactor.drain; + assert.deepStrictEqual(yield* Ref.get(fixture.removedWorktrees), []); + } + const session = { ...second.session!, status: "stopped" as const }; + yield* Ref.update(fixture.snapshots, (snapshot) => ({ + ...snapshot, + threads: snapshot.threads.map((thread) => + thread.id === second.id ? { ...thread, session } : thread, + ), + })); + yield* fixture.publishEvent({ + ...settledEvent(second.id), + type: "thread.session-set", + payload: { threadId: second.id, session }, + }); + yield* Queue.take(fixture.snapshotReads); + yield* reactor.drain; + assert.deepStrictEqual(yield* Ref.get(fixture.removedWorktrees), [ + { cwd: "/workspace/project", path: worktreePath }, + ]); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + + it.effect("skips a settlement event when the thread has already been reactivated", () => + Effect.scoped( + Effect.gen(function* () { + const thread = makeThread("reopened", { worktreePath, settledOverride: "active" }); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([thread]), + settings: cleanupSettings, + }); + yield* Effect.gen(function* () { + const reactor = yield* ThreadSettlementReactor.ThreadSettlementReactor; + yield* reactor.start(); + yield* Deferred.succeed(fixture.activation, undefined); + yield* fixture.publishEvent(settledEvent(thread.id)); + yield* Queue.take(fixture.snapshotReads); + yield* reactor.drain; + assert.deepStrictEqual(yield* Ref.get(fixture.removedWorktrees), []); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + + it.effect("continues cleanup after Git refuses a worktree removal", () => + Effect.scoped( + Effect.gen(function* () { + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([ + makeThread("dirty", { worktreePath }), + makeThread("clean", { worktreePath: "/workspace/clean" }), + ]), + settings: cleanupSettings, + removeWorktree: (input) => + input.path === worktreePath + ? Effect.fail( + new GitCommandError({ + operation: "removeWorktree", + cwd: input.cwd, + command: "git worktree remove", + detail: "worktree has changes", + }), + ) + : Effect.void, + }); + yield* Effect.gen(function* () { + const reactor = yield* ThreadSettlementReactor.ThreadSettlementReactor; + yield* reactor.start(); + yield* Deferred.succeed(fixture.activation, undefined); + for (const id of ["dirty", "clean"]) { + yield* fixture.settle(ThreadId.make(id)); + yield* Queue.take(fixture.snapshotReads); + yield* reactor.drain; + } + assert.deepStrictEqual(yield* Ref.get(fixture.removedWorktrees), [ + { cwd: "/workspace/project", path: worktreePath }, + { cwd: "/workspace/project", path: "/workspace/clean" }, + ]); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + it.effect( "settles all-terminal links from snapshots and keeps open or unsynced links active", () => diff --git a/apps/server/src/orchestration/ThreadSettlementReactor.ts b/apps/server/src/orchestration/ThreadSettlementReactor.ts index 6e262250de37..9c783e5f389e 100644 --- a/apps/server/src/orchestration/ThreadSettlementReactor.ts +++ b/apps/server/src/orchestration/ThreadSettlementReactor.ts @@ -1,5 +1,5 @@ import { normalizeSourceControlRepository } from "@t3tools/shared/sourceControl"; -import { CommandId } from "@t3tools/contracts"; +import { CommandId, type ThreadId } from "@t3tools/contracts"; import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; import * as Cause from "effect/Cause"; import * as Context from "effect/Context"; @@ -8,11 +8,13 @@ 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 Path from "effect/Path"; import * as Schedule from "effect/Schedule"; import type * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; import * as GitManager from "../git/GitManager.ts"; +import { GitWorkflowService } from "../git/GitWorkflowService.ts"; import * as PullRequestService from "../pullRequest/PullRequestService.ts"; import * as ServerSettings from "../serverSettings.ts"; import { forkParked } from "../serverActivation.ts"; @@ -42,6 +44,68 @@ export const make = Effect.gen(function* () { const pullRequests = yield* PullRequestService.PullRequestService; const crypto = yield* Crypto.Crypto; const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const gitWorkflow = yield* GitWorkflowService; + + const cleanupWorktree = Effect.fn("ThreadSettlementReactor.cleanupWorktree")(function* ( + threadId: ThreadId, + ) { + const snapshot = yield* snapshots.getShellSnapshot(); + const thread = snapshot.threads.find((thread) => thread.id === threadId); + if ( + thread === undefined || + thread.worktreePath === null || + (thread.settledOverride !== "settled" && thread.settledAt === null) || + (thread.session !== null && thread.session.status !== "stopped") + ) + return; + const projects = new Map(snapshot.projects.map((project) => [project.id, project])); + const project = projects.get(thread.projectId); + if (project === undefined) return; + + const canonicalPaths = new Map(); + const canonicalize = Effect.fn("ThreadSettlementReactor.canonicalize")(function* (cwd: string) { + const normalized = path.resolve(cwd); + const cached = canonicalPaths.get(normalized); + if (cached !== undefined) return cached; + const canonical = yield* fileSystem + .realPath(normalized) + .pipe(Effect.orElseSucceed(() => normalized)); + canonicalPaths.set(normalized, canonical); + return canonical; + }); + const worktreePath = yield* canonicalize(thread.worktreePath); + if (worktreePath === (yield* canonicalize(project.workspaceRoot))) return; + + for (const other of snapshot.threads) { + if (other.id === threadId || other.archivedAt !== null) continue; + // A settled thread can still be shutting down its provider session. + if ( + (other.settledOverride === "settled" || other.settledAt !== null) && + (other.session === null || other.session.status === "stopped") + ) + continue; + const cwd = other.worktreePath ?? projects.get(other.projectId)?.workspaceRoot; + if (cwd !== undefined && (yield* canonicalize(cwd)) === worktreePath) return; + } + + // Git preserves the branch and refuses to remove a main or dirty worktree. + yield* gitWorkflow.removeWorktree({ cwd: project.workspaceRoot, path: worktreePath }); + yield* git.invalidateStatus(project.workspaceRoot); + yield* git.invalidateStatus(worktreePath); + }); + const cleanupWorker = yield* makeDrainableWorker((threadId: ThreadId) => + cleanupWorktree(threadId).pipe( + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.failCause(cause) + : Effect.logWarning("settled thread worktree cleanup failed", { + threadId, + cause: Cause.pretty(cause), + }), + ), + ), + ); const sweep = Effect.fn("ThreadSettlementReactor.sweep")(function* ( mergedPullRequest: PullRequestService.PullRequestMergeEvent | null, @@ -253,6 +317,15 @@ export const make = Effect.gen(function* () { const start: ThreadSettlementReactor["Service"]["start"] = Effect.fn( "ThreadSettlementReactor.start", )(function* () { + const events = yield* engine.subscribeDomainEvents; + yield* forkParked( + Stream.runForEach(events, (event) => + event.type === "thread.settled" || + (event.type === "thread.session-set" && event.payload.session.status === "stopped") + ? cleanupWorker.enqueue(event.payload.threadId) + : Effect.void, + ), + ); const settingsChanges = yield* settingsService.subscribeChanges; const mergedPullRequests = yield* pullRequests.subscribeMerges; const initialSettings = yield* settingsService.getSettings.pipe(Effect.orDie); @@ -280,7 +353,10 @@ export const make = Effect.gen(function* () { yield* forkParked(Stream.runForEach(mergedPullRequests, runSweep)); }); - return { start, drain: worker.drain } satisfies ThreadSettlementReactor["Service"]; + return { + start, + drain: worker.drain.pipe(Effect.andThen(cleanupWorker.drain)), + } satisfies ThreadSettlementReactor["Service"]; }); export const layer = Layer.effect(ThreadSettlementReactor, make); diff --git a/apps/server/src/vcs/GitVcsDriverCore.test.ts b/apps/server/src/vcs/GitVcsDriverCore.test.ts index fb741b28ee7c..a00dd9ff1565 100644 --- a/apps/server/src/vcs/GitVcsDriverCore.test.ts +++ b/apps/server/src/vcs/GitVcsDriverCore.test.ts @@ -1567,6 +1567,7 @@ it.layer(TestLayer)("GitVcsDriver core integration", (it) => { yield* driver.removeWorktree({ cwd, path: worktreePath }); const fileSystem = yield* FileSystem.FileSystem; assert.equal(yield* fileSystem.exists(worktreePath), false); + assert.equal(yield* git(cwd, ["branch", "--list", "feature/worktree"]), "feature/worktree"); }), ); diff --git a/docs/user/thread-sidebar.md b/docs/user/thread-sidebar.md index 360057dd56a1..2fae828469cb 100644 --- a/docs/user/thread-sidebar.md +++ b/docs/user/thread-sidebar.md @@ -78,6 +78,10 @@ and prevents automatic settlement until new activity resumes the usual rules. Manually settling an idle thread dismisses unanswered async questions without sending an answer or restarting the agent. +Settling also removes the thread's worktree once no active threads use it. The main +worktree, branches, and worktrees with uncommitted changes are preserved. Starting +a new turn in a settled thread recreates a removed worktree from its saved branch. + By default, environments settle inactive threads after three days and settle threads whose pull request merged. A closed pull request can also settle an idle thread. Work in progress, pending questions or approvals, and live background work