From 7ac952c630ef8a60d14ca8f6a0c7adce33db4bae Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Wed, 14 Oct 2026 05:56:56 -0700 Subject: [PATCH] fix(server): worktree cleanup no longer runs sweeps back to back (#19418) Co-authored-by: Tanner Sharon Co-authored-by: Claude Opus 5.5 --- .../orchestration-v2/ProjectionStore.test.ts | 63 +++++++ .../src/orchestration-v2/ProjectionStore.ts | 39 ++++ .../ProviderTurnControlService.test.ts | 1 + apps/server/src/storageCleanup.test.ts | 176 ++++++++++++++---- apps/server/src/storageCleanup.ts | 76 +++++--- 5 files changed, 296 insertions(+), 59 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 8449e06cb6537a2b735c87da7822ec013a2d5b08..d2bd4a13ffecda56e7a53c2a828f8f54b23bcd95 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -327,6 +327,65 @@ const restartCancelledWorkSurvivesStaleRunUpdate = Effect.gen(function* () { assert.deepEqual(updated?.restartCancelledBackgroundWork, work); }); +const threadWorktreePathsReadActiveThreads = Effect.gen(function* () { + const store = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const create = (id: string, input: { worktreePath: string | null; archived?: boolean }) => { + const threadId = ThreadId.make(id); + const thread = { + createdBy: "user" as const, + creationSource: "web" as const, + id: threadId, + projectId: ProjectId.make("project:worktree-paths"), + title: id, + providerInstanceId, + modelSelection, + runtimeMode: "full-access" as const, + interactionMode: "default" as const, + branch: input.worktreePath === null ? null : id, + worktreePath: input.worktreePath, + activeProviderThreadId: null, + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: input.archived === true ? now : null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }; + return store + .apply({ + id: EventId.make(`event:${id}:created`), + type: "thread.created", + threadId, + occurredAt: now, + payload: thread, + }) + .pipe(Effect.as(thread)); + }; + yield* create("thread:active", { worktreePath: "/worktrees/active" }); + yield* create("thread:archived", { worktreePath: "/worktrees/archived", archived: true }); + yield* create("thread:local", { worktreePath: null }); + const deleted = yield* create("thread:deleted", { worktreePath: "/worktrees/deleted" }); + yield* store.apply({ + id: EventId.make("event:thread:deleted:deleted"), + type: "thread.deleted", + threadId: deleted.id, + occurredAt: now, + payload: { ...deleted, deletedAt: now }, + }); + + assert.deepEqual(yield* store.getThreadWorktreePaths(), [ + { threadId: ThreadId.make("thread:active"), worktreePath: "/worktrees/active" }, + ]); +}); + +it.effect("memory worktree path lookup reads only active threads", () => + threadWorktreePathsReadActiveThreads.pipe(Effect.provide(ProjectionStore.layerMemory)), +); + it.effect("memory projection keeps restart-cancelled work through a stale run.updated", () => restartCancelledWorkSurvivesStaleRunUpdate.pipe(Effect.provide(ProjectionStore.layerMemory)), ); @@ -553,6 +612,10 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { "SQL recovery selects unfinished runless native subagents", () => runlessNativeSubagentNeedsRecovery, ); + it.effect( + "worktree path lookup reads only active threads", + () => threadWorktreePathsReadActiveThreads, + ); it.effect( "keeps restart-cancelled work through a stale run.updated", diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 44c7c79579174305b66c310287907c36732876f7..7443eb7aab8d102065bc0a0278048bf8bfb16e26 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -416,6 +416,15 @@ export interface ProjectionStoreV2Shape { readonly getThreadsWithPullRequests: ( threadId?: ThreadId, ) => Effect.Effect, ProjectionStoreV2Error>; + /** + * Active (not deleted, not archived) threads that have a worktree path, the + * same threads as the shell snapshot's `threads`. Reads only the thread rows, + * so callers can check who uses a path without a shell snapshot. + */ + readonly getThreadWorktreePaths: () => Effect.Effect< + ReadonlyArray<{ readonly threadId: ThreadId; readonly worktreePath: string }>, + ProjectionStoreV2Error + >; readonly getTurnStartContext: ( threadId: ThreadId, runId: RunId, @@ -5699,6 +5708,23 @@ export const layer: Layer.Layer = ); }).pipe(Effect.mapError((cause) => new ProjectionStoreSetupError({ cause }))); + const getThreadWorktreePaths: ProjectionStoreV2Shape["getThreadWorktreePaths"] = () => + sql<{ readonly thread_id: string; readonly worktree_path: string }>` + SELECT thread_id, json_extract(payload_json, '$.worktreePath') AS worktree_path + FROM orchestration_v2_projection_threads + WHERE deleted_at IS NULL + AND json_extract(payload_json, '$.archivedAt') IS NULL + AND json_extract(payload_json, '$.worktreePath') IS NOT NULL + `.pipe( + Effect.map((rows) => + rows.map((row) => ({ + threadId: ThreadId.make(row.thread_id), + worktreePath: row.worktree_path, + })), + ), + Effect.mapError((cause) => new ProjectionStoreSetupError({ cause })), + ); + const shellThreadStateFromRow = (input: { readonly row: ShellThreadRow; readonly runOrdinalsByThreadId: ReadonlyMap>; @@ -6035,6 +6061,7 @@ export const layer: Layer.Layer = getThread, getSettlementCandidates, getThreadsWithPullRequests, + getThreadWorktreePaths, getThreadProjection, getTurnStartContext, getTurnStartHistory, @@ -6184,6 +6211,18 @@ export const layerMemory: Layer.Layer = Layer.effect( left.id.localeCompare(right.id), ); }), + getThreadWorktreePaths: () => + Ref.get(replayState).pipe( + Effect.map((state) => + [...state.projections.values()].flatMap(({ thread }) => + thread.deletedAt === null && + thread.archivedAt === null && + thread.worktreePath !== null + ? [{ threadId: thread.id, worktreePath: thread.worktreePath }] + : [], + ), + ), + ), getThreadsWithPullRequests: (threadId) => Ref.get(replayState).pipe( Effect.map((state) => diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index cafd318526434a9f35ed520d4008def832b0bf75..fc3465e6885f1595430727548150e1238fcd972e 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -218,6 +218,7 @@ it.effect( getThread: () => Ref.get(projection).pipe(Effect.map((state) => state.thread)), getSettlementCandidates: () => Effect.die("unused getSettlementCandidates"), getThreadsWithPullRequests: () => Effect.die("unused getThreadsWithPullRequests"), + getThreadWorktreePaths: () => Effect.die("unused getThreadWorktreePaths"), getThreadProjection: () => Effect.die("control effects must not load transcript"), getTurnStartContext: () => Effect.die("unused"), getTurnStartHistory: () => Effect.die("unused"), diff --git a/apps/server/src/storageCleanup.test.ts b/apps/server/src/storageCleanup.test.ts index 5488b366c246470d7ba7bc9c41af74c32330106b..2b5c1f69400615b90e9b38645127fac7b9cdb98f 100644 --- a/apps/server/src/storageCleanup.test.ts +++ b/apps/server/src/storageCleanup.test.ts @@ -28,11 +28,16 @@ import { CodexProviderCapabilitiesV2 } from "./orchestration-v2/Adapters/CodexAd import { ServerActivation } from "./serverActivation.ts"; import { describe, expect, it } from "@effect/vitest"; import { + EventId, ProjectId, + ProviderDriverKind, ProviderInstanceId, + ProviderSessionId, RunId, RuntimeRequestId, ThreadId, + type OrchestrationV2DomainEvent, + type OrchestrationV2ProviderSession, type OrchestrationV2ThreadShell, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; @@ -243,14 +248,22 @@ const cleanupFixture = Effect.gen(function* () { yield* fs.makeDirectory(config.worktreesDir, { recursive: true }); const worktree = `${config.worktreesDir}/feature`; yield* command(repo, ["worktree", "add", "-b", "feature", worktree]); + // worktreeOnDelete lets domain events request sweeps once a test calls activate. let settings: import("@t3tools/contracts").ServerSettings = { ...DEFAULT_SERVER_SETTINGS, - storageCleanup: { ...DEFAULT_SERVER_SETTINGS.storageCleanup, worktreeAfterDays: 8 }, + storageCleanup: { + ...DEFAULT_SERVER_SETTINGS.storageCleanup, + worktreeAfterDays: 8, + worktreeOnDelete: true, + }, }; let threads = [shell({ branch: "feature", worktreePath: worktree })]; let editBeforeRemoval = false; let openSessionCwd: string | null = null; let fullStatusCalls = 0; + let snapshotReads = 0; + let events: Stream.Stream = Stream.empty; + const activated = yield* Deferred.make(); const context = yield* Layer.build(StorageCleanup.layer).pipe( Effect.provideService(GitVcsDriver.GitVcsDriver, { ...git, @@ -274,10 +287,23 @@ const cleanupFixture = Effect.gen(function* () { } as unknown as ProjectStore.ProjectStoreV2["Service"]), Effect.provideService(ProjectionStore.ProjectionStoreV2, { getShellSnapshot: (input?: { location?: string }) => - Effect.sync(() => ({ threads: input?.location === "archive" ? [] : threads })), + Effect.sync(() => { + snapshotReads++; + return { threads: input?.location === "archive" ? [] : threads }; + }), + getThreadWorktreePaths: () => + Effect.sync(() => + threads.flatMap((thread) => + thread.worktreePath === null + ? [] + : [{ threadId: thread.id, worktreePath: thread.worktreePath }], + ), + ), + getThreadShell: (threadId: ThreadId) => + Effect.sync(() => threads.find((thread) => thread.id === threadId) ?? null), } as unknown as ProjectionStore.ProjectionStoreV2["Service"]), Effect.provideService(Orchestrator.OrchestratorV2, { - streamDomainEvents: Stream.empty, + streamDomainEvents: Stream.suspend(() => events), } as unknown as Orchestrator.OrchestratorV2["Service"]), Effect.provideService(SqlClient.SqlClient, ((strings: TemplateStringsArray) => Effect.sync(() => @@ -306,7 +332,7 @@ const cleanupFixture = Effect.gen(function* () { Effect.provideService(TerminalManager.TerminalManager, { subscribeMetadata: () => Effect.succeed(() => {}), } as unknown as TerminalManager.TerminalManager["Service"]), - Effect.provideService(ServerActivation, Effect.never), + Effect.provideService(ServerActivation, Deferred.await(activated)), ); return { service: Context.get(context, StorageCleanup.StorageCleanup), @@ -333,7 +359,11 @@ const cleanupFixture = Effect.gen(function* () { disable: () => { settings = { ...settings, - storageCleanup: { ...settings.storageCleanup, worktreeAfterDays: null }, + storageCleanup: { + ...settings.storageCleanup, + worktreeAfterDays: null, + worktreeOnDelete: false, + }, }; }, setThreads: (next: typeof threads) => { @@ -343,8 +373,57 @@ const cleanupFixture = Effect.gen(function* () { openSessionCwd = cwd; }, fullStatusCalls: () => fullStatusCalls, + snapshotReads: () => snapshotReads, + /** Starts the hourly sweep and the subscription to `domainEvents`, which wait until now. */ + activate: (domainEvents: Stream.Stream) => { + events = domainEvents; + return Deferred.succeed(activated, undefined); + }, }; }); +/** The fixture, with the size walk of its worktree held until `resume` succeeds. */ +const pausedMeasurementFixture = Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const measuring = yield* Deferred.make(); + const resume = yield* Deferred.make(); + let measuredPath: string | undefined; + const fixture = yield* cleanupFixture.pipe( + Effect.provideService(FileSystem.FileSystem, { + ...fs, + readDirectory: (directory, options) => + Effect.gen(function* () { + if (directory === measuredPath) { + yield* Deferred.succeed(measuring, undefined); + yield* Deferred.await(resume); + } + return yield* fs.readDirectory(directory, options); + }), + }), + ); + measuredPath = yield* fs.realPath(fixture.worktree); + return { ...fixture, measuring, resume }; +}); +const sessionUpdated = ( + index: number, + status: OrchestrationV2ProviderSession["status"], +): OrchestrationV2DomainEvent => ({ + id: EventId.make(`event-${index}`), + type: "provider-session.updated", + threadId: ThreadId.make("thread-1"), + occurredAt: at(0), + payload: { + id: ProviderSessionId.make(`session-${index}`), + driver: ProviderDriverKind.make("codex"), + providerInstanceId: ProviderInstanceId.make("codex"), + status, + cwd: "/elsewhere", + model: null, + capabilities: CodexProviderCapabilitiesV2, + createdAt: at(0), + updatedAt: at(0), + lastError: null, + }, +}); const cleanupTestLayer = GitVcsDriver.layer.pipe( Layer.provideMerge(ServerConfig.layerTest(process.cwd(), { prefix: "storage-cleanup-" })), Layer.provideMerge(NodeServices.layer), @@ -402,35 +481,31 @@ describe("storage cleanup reports and local file policies", () => { }), ), ); - it.live("keeps a worktree when its thread starts during size measurement", () => + it.live.each([ + [ + "its thread starts", + (worktree: string) => [ + shell({ branch: "feature", worktreePath: worktree, status: "running" }), + ], + ], + [ + "another thread starts using it", + (worktree: string) => [ + shell({ branch: "feature", worktreePath: worktree }), + shell({ id: ThreadId.make("thread-2"), branch: "feature", worktreePath: worktree }), + ], + ], + ] as const)("keeps a worktree when %s during size measurement", ([, threadsWhileMeasuring]) => runCleanupTest( Effect.gen(function* () { - const fs = yield* FileSystem.FileSystem; - const measuring = yield* Deferred.make(); - const resume = yield* Deferred.make(); - let measuredPath: string | undefined; - const fixture = yield* cleanupFixture.pipe( - Effect.provideService(FileSystem.FileSystem, { - ...fs, - readDirectory: (directory) => - Effect.gen(function* () { - if (directory === measuredPath) { - yield* Deferred.succeed(measuring, undefined); - yield* Deferred.await(resume); - } - return yield* fs.readDirectory(directory); - }), - }), - ); - measuredPath = yield* fs.realPath(fixture.worktree); + const fixture = yield* pausedMeasurementFixture; + const { fs, worktree } = fixture; fixture.setPolicy("tracked-changes"); - yield* fs.writeFileString(`${fixture.worktree}/notes.txt`, "untracked\n"); + yield* fs.writeFileString(`${worktree}/notes.txt`, "untracked\n"); const cleanup = yield* fixture.service.runNow.pipe(Effect.forkChild); - yield* Deferred.await(measuring); - fixture.setThreads([ - shell({ branch: "feature", worktreePath: fixture.worktree, status: "running" }), - ]); - yield* Deferred.succeed(resume, undefined); + yield* Deferred.await(fixture.measuring); + fixture.setThreads(threadsWhileMeasuring(worktree)); + yield* Deferred.succeed(fixture.resume, undefined); const report = yield* Fiber.join(cleanup); expect(report.entries[0]).toMatchObject({ outcome: "kept", @@ -438,11 +513,48 @@ describe("storage cleanup reports and local file policies", () => { bytes: null, }); expect(report.bytesFreed).toBe(0); - expect(yield* fs.exists(fixture.worktree)).toBe(true); - expect(yield* fs.readFileString(`${fixture.worktree}/notes.txt`)).toBe("untracked\n"); + expect(yield* fs.exists(worktree)).toBe(true); + expect(yield* fs.readFileString(`${worktree}/notes.txt`)).toBe("untracked\n"); }), ), ); + it.live("reads the thread snapshot once per sweep, not per worktree", () => + runCleanupTest( + Effect.gen(function* () { + const { service, snapshotReads } = yield* cleanupFixture; + expect((yield* service.runNow).entries[0]).toMatchObject({ outcome: "removed" }); + expect(snapshotReads()).toBe(1); + }), + ), + ); + it.live.each([ + ["running", "ready", 1], + ["stopped", "error", 2], + ] as const)( + "session updates to %s or %s during a sweep make %i sweep(s) in total", + ([first, second, sweeps]) => + runCleanupTest( + Effect.gen(function* () { + const fixture = yield* pausedMeasurementFixture; + const handled = yield* Deferred.make(); + // 20 updates arrive while the first automatic sweep measures the worktree. + yield* fixture.activate( + Stream.fromEffect(Deferred.await(fixture.measuring)).pipe( + Stream.flatMap(() => Stream.range(0, 19)), + Stream.map((index) => sessionUpdated(index, index % 2 === 0 ? first : second)), + Stream.concat( + Stream.fromEffect(Deferred.succeed(handled, undefined)).pipe(Stream.drain), + ), + ), + ); + yield* Deferred.await(handled); + yield* Deferred.succeed(fixture.resume, undefined); + yield* fixture.service.drain; + // One snapshot read per sweep. + expect(fixture.snapshotReads()).toBe(sweeps); + }), + ), + ); it.live("does not measure a worktree kept for an open provider session", () => runCleanupTest( Effect.gen(function* () { diff --git a/apps/server/src/storageCleanup.ts b/apps/server/src/storageCleanup.ts index 60b37521c0d0feedcd9aca460c712e69195946a9..20f1ed9f6ed92f83c226f1658fdde6ee4ab71c95 100644 --- a/apps/server/src/storageCleanup.ts +++ b/apps/server/src/storageCleanup.ts @@ -27,6 +27,7 @@ import * as Equal from "effect/Equal"; import * as FileSystem from "effect/FileSystem"; import * as Option from "effect/Option"; import * as Path from "effect/Path"; +import * as Ref from "effect/Ref"; import type { PlatformError } from "effect/PlatformError"; import * as Schedule from "effect/Schedule"; import * as Stream from "effect/Stream"; @@ -45,6 +46,7 @@ import { isFilesystemRoot, managedWorktreesDirectories } from "./worktreesDirect import * as GitVcsDriver from "./vcs/GitVcsDriver.ts"; import { withWorkspaceLease } from "./workspace/workspaceLease.ts"; import * as HostProcess from "@t3tools/shared/HostProcess"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; const decodeCleanupThread = Schema.decodeUnknownEffect( Schema.fromJsonString(OrchestrationV2AppThreadJson), @@ -248,11 +250,16 @@ const make = Effect.gen(function* () { Effect.orElseSucceed(() => null), ); + // Active threads only: archived threads are not candidates and do not count + // as sharing a worktree. The recheck's getThreadWorktreePaths reads the same set. + // The snapshot takes seconds on large databases, so it runs on the reader + // worker instead of the event loop. const readThreads = Effect.fn("StorageCleanup.readThreads")(function* () { - const active = yield* projections.getShellSnapshot(); - const archived = yield* projections.getShellSnapshot({ location: "archive" }); + const { threads } = yield* projections + .getShellSnapshot({ location: "active" }) + .pipe(NodeSqliteClient.readOnly); const projects = yield* projectStore.listShells(); - return { projects, threads: [...active.threads, ...archived.threads] }; + return { projects, threads }; }); // Local threads under another project need not have a worktreePath of their own. @@ -538,14 +545,14 @@ const make = Effect.gen(function* () { ); // Threads, queued deletion work and provider sessions as they are now, so a // queued turn, resumed session or new thread sharing this path cancels - // the removal. + // the removal. Read only this path and this thread: a full snapshot per + // worktree takes seconds on large databases. const blockedSinceCheck = Effect.fnUntraced(function* () { - const latestSnapshot = yield* readThreads(); - if (yield* containsProjectRoot(worktreePath, [project, ...latestSnapshot.projects])) + const latestProjects = yield* projectStore.listShells(); + if (yield* containsProjectRoot(worktreePath, [project, ...latestProjects])) return "contains a project checkout"; - const latest = latestSnapshot.threads.filter( - (entry) => - entry.worktreePath !== null && path.resolve(entry.worktreePath) === worktreePath, + const latest = (yield* projections.getThreadWorktreePaths()).filter( + (entry) => path.resolve(entry.worktreePath) === worktreePath, ); if (hasTerminal(worktreePath)) return "open terminal"; if (deleted) { @@ -562,13 +569,18 @@ const make = Effect.gen(function* () { WHERE thread_id = ${thread.id} AND status NOT IN ('succeeded', 'cancelled') LIMIT 1 `; if (pendingCleanup.length > 0) return "thread deletion is still pending"; - } else if ( - latest.length !== 1 || - latest[0]!.id !== thread.id || - !storageCleanupThreadIdle(latest[0]!, now) || - storageCleanupActivityAt(latest[0]!) !== storageCleanupActivityAt(thread) - ) - return "thread activity or shared worktree changed since check"; + } else { + const latestShell = + latest.length === 1 && latest[0]!.threadId === thread.id + ? yield* projections.getThreadShell(thread.id) + : null; + if ( + latestShell === null || + !storageCleanupThreadIdle(latestShell, now) || + storageCleanupActivityAt(latestShell) !== storageCleanupActivityAt(thread) + ) + return "thread activity or shared worktree changed since check"; + } // Sessions can outlive their run and can be shared across app threads. const sessionRows = yield* sql<{ payload_json: string }>` SELECT payload_json FROM orchestration_v2_projection_provider_sessions @@ -775,9 +787,14 @@ const make = Effect.gen(function* () { yield* SubscriptionRef.set(reportRef, latestReport); return latestReport; }); + // At most one automatic sweep waits at a time. An automatic request made + // while one is waiting joins it, since that sweep has not read any state yet. + // Each manual run still gets its own sweep and report. + const automaticSweepQueued = yield* Ref.make(false); const worker = yield* makeDrainableWorker( (completion: Deferred.Deferred | undefined) => - sweep(completion === undefined ? "automatic" : "manual").pipe( + (completion === undefined ? Ref.set(automaticSweepQueued, false) : Effect.void).pipe( + Effect.andThen(sweep(completion === undefined ? "automatic" : "manual")), Effect.exit, Effect.flatMap((exit) => completion === undefined @@ -791,6 +808,9 @@ const make = Effect.gen(function* () { ), ), ); + const requestSweep = Ref.getAndSet(automaticSweepQueued, true).pipe( + Effect.flatMap((queued) => (queued ? Effect.void : worker.enqueue(undefined))), + ); const runNow = Effect.gen(function* () { const completion = yield* Deferred.make(); yield* worker.enqueue(completion); @@ -817,13 +837,11 @@ const make = Effect.gen(function* () { const events = engine.streamDomainEvents; let lastSettings = yield* settingsService.getSettings.pipe(Effect.orDie); yield* forkParked( - worker - .enqueue(undefined) - .pipe( - Effect.andThen(worker.drain), - Effect.repeat(Schedule.spaced("1 hour")), - Effect.asVoid, - ), + requestSweep.pipe( + Effect.andThen(worker.drain), + Effect.repeat(Schedule.spaced("1 hour")), + Effect.asVoid, + ), ); yield* forkParked( Stream.runForEach(changes, (settings) => { @@ -834,14 +852,18 @@ const make = Effect.gen(function* () { ) return Effect.void; lastSettings = settings; - return worker.enqueue(undefined); + return requestSweep; }), ); yield* forkParked( + // A deleted thread's worktree waits for its sessions to end. Moving + // between live statuses, such as running to ready, cannot free one. Stream.runForEach(events, (event) => - (event.type === "thread.deleted" || event.type === "provider-session.updated") && + (event.type === "thread.deleted" || + (event.type === "provider-session.updated" && + (event.payload.status === "stopped" || event.payload.status === "error"))) && anyWorktreePolicy(lastSettings, (rules) => rules.worktreeOnDelete) - ? worker.enqueue(undefined) + ? requestSweep : Effect.void, ).pipe( Effect.catchCause((cause) => -- 2.54.0