authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-10-14 05:56:56-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-14 05:56:56-07:00
log7ac952c630ef8a60d14ca8f6a0c7adce33db4bae
treec8a8e63c7405acb4e0cc3cbbc3e9cf73f534fa9d
parentd0a73b56c21ca1b7f39ab28d0fe46db99eec8f67
signature Signed by PGP key B5690EEEBB952194

fix(server): worktree cleanup no longer runs sweeps back to back (#19418)

Co-authored-by: Tanner Sharon <tannermsharon@gmail.com> Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

5 files changed, 296 insertions(+), 59 deletions(-)

apps/server/src/orchestration-v2/ProjectionStore.test.ts+63
...@@ -327,6 +327,65 @@ const restartCancelledWorkSurvivesStaleRunUpdate = Effect.gen(function* () {...@@ -327,6 +327,65 @@ const restartCancelledWorkSurvivesStaleRunUpdate = Effect.gen(function* () {
327 assert.deepEqual(updated?.restartCancelledBackgroundWork, work);327 assert.deepEqual(updated?.restartCancelledBackgroundWork, work);
328});328});
329329
330const threadWorktreePathsReadActiveThreads = Effect.gen(function* () {
331 const store = yield* ProjectionStore.ProjectionStoreV2;
332 const now = yield* DateTime.now;
333 const create = (id: string, input: { worktreePath: string | null; archived?: boolean }) => {
334 const threadId = ThreadId.make(id);
335 const thread = {
336 createdBy: "user" as const,
337 creationSource: "web" as const,
338 id: threadId,
339 projectId: ProjectId.make("project:worktree-paths"),
340 title: id,
341 providerInstanceId,
342 modelSelection,
343 runtimeMode: "full-access" as const,
344 interactionMode: "default" as const,
345 branch: input.worktreePath === null ? null : id,
346 worktreePath: input.worktreePath,
347 activeProviderThreadId: null,
348 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
349 forkedFrom: null,
350 createdAt: now,
351 updatedAt: now,
352 archivedAt: input.archived === true ? now : null,
353 settledOverride: null,
354 settledAt: null,
355 lastVisitedAt: null,
356 deletedAt: null,
357 };
358 return store
359 .apply({
360 id: EventId.make(`event:${id}:created`),
361 type: "thread.created",
362 threadId,
363 occurredAt: now,
364 payload: thread,
365 })
366 .pipe(Effect.as(thread));
367 };
368 yield* create("thread:active", { worktreePath: "/worktrees/active" });
369 yield* create("thread:archived", { worktreePath: "/worktrees/archived", archived: true });
370 yield* create("thread:local", { worktreePath: null });
371 const deleted = yield* create("thread:deleted", { worktreePath: "/worktrees/deleted" });
372 yield* store.apply({
373 id: EventId.make("event:thread:deleted:deleted"),
374 type: "thread.deleted",
375 threadId: deleted.id,
376 occurredAt: now,
377 payload: { ...deleted, deletedAt: now },
378 });
379
380 assert.deepEqual(yield* store.getThreadWorktreePaths(), [
381 { threadId: ThreadId.make("thread:active"), worktreePath: "/worktrees/active" },
382 ]);
383});
384
385it.effect("memory worktree path lookup reads only active threads", () =>
386 threadWorktreePathsReadActiveThreads.pipe(Effect.provide(ProjectionStore.layerMemory)),
387);
388
330it.effect("memory projection keeps restart-cancelled work through a stale run.updated", () =>389it.effect("memory projection keeps restart-cancelled work through a stale run.updated", () =>
331 restartCancelledWorkSurvivesStaleRunUpdate.pipe(Effect.provide(ProjectionStore.layerMemory)),390 restartCancelledWorkSurvivesStaleRunUpdate.pipe(Effect.provide(ProjectionStore.layerMemory)),
332);391);
...@@ -553,6 +612,10 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {...@@ -553,6 +612,10 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
553 "SQL recovery selects unfinished runless native subagents",612 "SQL recovery selects unfinished runless native subagents",
554 () => runlessNativeSubagentNeedsRecovery,613 () => runlessNativeSubagentNeedsRecovery,
555 );614 );
615 it.effect(
616 "worktree path lookup reads only active threads",
617 () => threadWorktreePathsReadActiveThreads,
618 );
556619
557 it.effect(620 it.effect(
558 "keeps restart-cancelled work through a stale run.updated",621 "keeps restart-cancelled work through a stale run.updated",
apps/server/src/orchestration-v2/ProjectionStore.ts+39
...@@ -416,6 +416,15 @@ export interface ProjectionStoreV2Shape {...@@ -416,6 +416,15 @@ export interface ProjectionStoreV2Shape {
416 readonly getThreadsWithPullRequests: (416 readonly getThreadsWithPullRequests: (
417 threadId?: ThreadId,417 threadId?: ThreadId,
418 ) => Effect.Effect<ReadonlyArray<ProjectionThreadPullRequests>, ProjectionStoreV2Error>;418 ) => Effect.Effect<ReadonlyArray<ProjectionThreadPullRequests>, ProjectionStoreV2Error>;
419 /**
420 * Active (not deleted, not archived) threads that have a worktree path, the
421 * same threads as the shell snapshot's `threads`. Reads only the thread rows,
422 * so callers can check who uses a path without a shell snapshot.
423 */
424 readonly getThreadWorktreePaths: () => Effect.Effect<
425 ReadonlyArray<{ readonly threadId: ThreadId; readonly worktreePath: string }>,
426 ProjectionStoreV2Error
427 >;
419 readonly getTurnStartContext: (428 readonly getTurnStartContext: (
420 threadId: ThreadId,429 threadId: ThreadId,
421 runId: RunId,430 runId: RunId,
...@@ -5699,6 +5708,23 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =...@@ -5699,6 +5708,23 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
5699 );5708 );
5700 }).pipe(Effect.mapError((cause) => new ProjectionStoreSetupError({ cause })));5709 }).pipe(Effect.mapError((cause) => new ProjectionStoreSetupError({ cause })));
57015710
5711 const getThreadWorktreePaths: ProjectionStoreV2Shape["getThreadWorktreePaths"] = () =>
5712 sql<{ readonly thread_id: string; readonly worktree_path: string }>`
5713 SELECT thread_id, json_extract(payload_json, '$.worktreePath') AS worktree_path
5714 FROM orchestration_v2_projection_threads
5715 WHERE deleted_at IS NULL
5716 AND json_extract(payload_json, '$.archivedAt') IS NULL
5717 AND json_extract(payload_json, '$.worktreePath') IS NOT NULL
5718 `.pipe(
5719 Effect.map((rows) =>
5720 rows.map((row) => ({
5721 threadId: ThreadId.make(row.thread_id),
5722 worktreePath: row.worktree_path,
5723 })),
5724 ),
5725 Effect.mapError((cause) => new ProjectionStoreSetupError({ cause })),
5726 );
5727
5702 const shellThreadStateFromRow = (input: {5728 const shellThreadStateFromRow = (input: {
5703 readonly row: ShellThreadRow;5729 readonly row: ShellThreadRow;
5704 readonly runOrdinalsByThreadId: ReadonlyMap<ThreadId, Map<RunId, number>>;5730 readonly runOrdinalsByThreadId: ReadonlyMap<ThreadId, Map<RunId, number>>;
...@@ -6035,6 +6061,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =...@@ -6035,6 +6061,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
6035 getThread,6061 getThread,
6036 getSettlementCandidates,6062 getSettlementCandidates,
6037 getThreadsWithPullRequests,6063 getThreadsWithPullRequests,
6064 getThreadWorktreePaths,
6038 getThreadProjection,6065 getThreadProjection,
6039 getTurnStartContext,6066 getTurnStartContext,
6040 getTurnStartHistory,6067 getTurnStartHistory,
...@@ -6184,6 +6211,18 @@ export const layerMemory: Layer.Layer<ProjectionStoreV2> = Layer.effect(...@@ -6184,6 +6211,18 @@ export const layerMemory: Layer.Layer<ProjectionStoreV2> = Layer.effect(
6184 left.id.localeCompare(right.id),6211 left.id.localeCompare(right.id),
6185 );6212 );
6186 }),6213 }),
6214 getThreadWorktreePaths: () =>
6215 Ref.get(replayState).pipe(
6216 Effect.map((state) =>
6217 [...state.projections.values()].flatMap(({ thread }) =>
6218 thread.deletedAt === null &&
6219 thread.archivedAt === null &&
6220 thread.worktreePath !== null
6221 ? [{ threadId: thread.id, worktreePath: thread.worktreePath }]
6222 : [],
6223 ),
6224 ),
6225 ),
6187 getThreadsWithPullRequests: (threadId) =>6226 getThreadsWithPullRequests: (threadId) =>
6188 Ref.get(replayState).pipe(6227 Ref.get(replayState).pipe(
6189 Effect.map((state) =>6228 Effect.map((state) =>
apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts+1
...@@ -218,6 +218,7 @@ it.effect(...@@ -218,6 +218,7 @@ it.effect(
218 getThread: () => Ref.get(projection).pipe(Effect.map((state) => state.thread)),218 getThread: () => Ref.get(projection).pipe(Effect.map((state) => state.thread)),
219 getSettlementCandidates: () => Effect.die("unused getSettlementCandidates"),219 getSettlementCandidates: () => Effect.die("unused getSettlementCandidates"),
220 getThreadsWithPullRequests: () => Effect.die("unused getThreadsWithPullRequests"),220 getThreadsWithPullRequests: () => Effect.die("unused getThreadsWithPullRequests"),
221 getThreadWorktreePaths: () => Effect.die("unused getThreadWorktreePaths"),
221 getThreadProjection: () => Effect.die("control effects must not load transcript"),222 getThreadProjection: () => Effect.die("control effects must not load transcript"),
222 getTurnStartContext: () => Effect.die("unused"),223 getTurnStartContext: () => Effect.die("unused"),
223 getTurnStartHistory: () => Effect.die("unused"),224 getTurnStartHistory: () => Effect.die("unused"),
apps/server/src/storageCleanup.test.ts+144-32
...@@ -28,11 +28,16 @@ import { CodexProviderCapabilitiesV2 } from "./orchestration-v2/Adapters/CodexAd...@@ -28,11 +28,16 @@ import { CodexProviderCapabilitiesV2 } from "./orchestration-v2/Adapters/CodexAd
28import { ServerActivation } from "./serverActivation.ts";28import { ServerActivation } from "./serverActivation.ts";
29import { describe, expect, it } from "@effect/vitest";29import { describe, expect, it } from "@effect/vitest";
30import {30import {
31 EventId,
31 ProjectId,32 ProjectId,
33 ProviderDriverKind,
32 ProviderInstanceId,34 ProviderInstanceId,
35 ProviderSessionId,
33 RunId,36 RunId,
34 RuntimeRequestId,37 RuntimeRequestId,
35 ThreadId,38 ThreadId,
39 type OrchestrationV2DomainEvent,
40 type OrchestrationV2ProviderSession,
36 type OrchestrationV2ThreadShell,41 type OrchestrationV2ThreadShell,
37} from "@t3tools/contracts";42} from "@t3tools/contracts";
38import * as DateTime from "effect/DateTime";43import * as DateTime from "effect/DateTime";
...@@ -243,14 +248,22 @@ const cleanupFixture = Effect.gen(function* () {...@@ -243,14 +248,22 @@ const cleanupFixture = Effect.gen(function* () {
243 yield* fs.makeDirectory(config.worktreesDir, { recursive: true });248 yield* fs.makeDirectory(config.worktreesDir, { recursive: true });
244 const worktree = `${config.worktreesDir}/feature`;249 const worktree = `${config.worktreesDir}/feature`;
245 yield* command(repo, ["worktree", "add", "-b", "feature", worktree]);250 yield* command(repo, ["worktree", "add", "-b", "feature", worktree]);
251 // worktreeOnDelete lets domain events request sweeps once a test calls activate.
246 let settings: import("@t3tools/contracts").ServerSettings = {252 let settings: import("@t3tools/contracts").ServerSettings = {
247 ...DEFAULT_SERVER_SETTINGS,253 ...DEFAULT_SERVER_SETTINGS,
248 storageCleanup: { ...DEFAULT_SERVER_SETTINGS.storageCleanup, worktreeAfterDays: 8 },254 storageCleanup: {
255 ...DEFAULT_SERVER_SETTINGS.storageCleanup,
256 worktreeAfterDays: 8,
257 worktreeOnDelete: true,
258 },
249 };259 };
250 let threads = [shell({ branch: "feature", worktreePath: worktree })];260 let threads = [shell({ branch: "feature", worktreePath: worktree })];
251 let editBeforeRemoval = false;261 let editBeforeRemoval = false;
252 let openSessionCwd: string | null = null;262 let openSessionCwd: string | null = null;
253 let fullStatusCalls = 0;263 let fullStatusCalls = 0;
264 let snapshotReads = 0;
265 let events: Stream.Stream<OrchestrationV2DomainEvent> = Stream.empty;
266 const activated = yield* Deferred.make<void>();
254 const context = yield* Layer.build(StorageCleanup.layer).pipe(267 const context = yield* Layer.build(StorageCleanup.layer).pipe(
255 Effect.provideService(GitVcsDriver.GitVcsDriver, {268 Effect.provideService(GitVcsDriver.GitVcsDriver, {
256 ...git,269 ...git,
...@@ -274,10 +287,23 @@ const cleanupFixture = Effect.gen(function* () {...@@ -274,10 +287,23 @@ const cleanupFixture = Effect.gen(function* () {
274 } as unknown as ProjectStore.ProjectStoreV2["Service"]),287 } as unknown as ProjectStore.ProjectStoreV2["Service"]),
275 Effect.provideService(ProjectionStore.ProjectionStoreV2, {288 Effect.provideService(ProjectionStore.ProjectionStoreV2, {
276 getShellSnapshot: (input?: { location?: string }) =>289 getShellSnapshot: (input?: { location?: string }) =>
277 Effect.sync(() => ({ threads: input?.location === "archive" ? [] : threads })),290 Effect.sync(() => {
291 snapshotReads++;
292 return { threads: input?.location === "archive" ? [] : threads };
293 }),
294 getThreadWorktreePaths: () =>
295 Effect.sync(() =>
296 threads.flatMap((thread) =>
297 thread.worktreePath === null
298 ? []
299 : [{ threadId: thread.id, worktreePath: thread.worktreePath }],
300 ),
301 ),
302 getThreadShell: (threadId: ThreadId) =>
303 Effect.sync(() => threads.find((thread) => thread.id === threadId) ?? null),
278 } as unknown as ProjectionStore.ProjectionStoreV2["Service"]),304 } as unknown as ProjectionStore.ProjectionStoreV2["Service"]),
279 Effect.provideService(Orchestrator.OrchestratorV2, {305 Effect.provideService(Orchestrator.OrchestratorV2, {
280 streamDomainEvents: Stream.empty,306 streamDomainEvents: Stream.suspend(() => events),
281 } as unknown as Orchestrator.OrchestratorV2["Service"]),307 } as unknown as Orchestrator.OrchestratorV2["Service"]),
282 Effect.provideService(SqlClient.SqlClient, ((strings: TemplateStringsArray) =>308 Effect.provideService(SqlClient.SqlClient, ((strings: TemplateStringsArray) =>
283 Effect.sync(() =>309 Effect.sync(() =>
...@@ -306,7 +332,7 @@ const cleanupFixture = Effect.gen(function* () {...@@ -306,7 +332,7 @@ const cleanupFixture = Effect.gen(function* () {
306 Effect.provideService(TerminalManager.TerminalManager, {332 Effect.provideService(TerminalManager.TerminalManager, {
307 subscribeMetadata: () => Effect.succeed(() => {}),333 subscribeMetadata: () => Effect.succeed(() => {}),
308 } as unknown as TerminalManager.TerminalManager["Service"]),334 } as unknown as TerminalManager.TerminalManager["Service"]),
309 Effect.provideService(ServerActivation, Effect.never),335 Effect.provideService(ServerActivation, Deferred.await(activated)),
310 );336 );
311 return {337 return {
312 service: Context.get(context, StorageCleanup.StorageCleanup),338 service: Context.get(context, StorageCleanup.StorageCleanup),
...@@ -333,7 +359,11 @@ const cleanupFixture = Effect.gen(function* () {...@@ -333,7 +359,11 @@ const cleanupFixture = Effect.gen(function* () {
333 disable: () => {359 disable: () => {
334 settings = {360 settings = {
335 ...settings,361 ...settings,
336 storageCleanup: { ...settings.storageCleanup, worktreeAfterDays: null },362 storageCleanup: {
363 ...settings.storageCleanup,
364 worktreeAfterDays: null,
365 worktreeOnDelete: false,
366 },
337 };367 };
338 },368 },
339 setThreads: (next: typeof threads) => {369 setThreads: (next: typeof threads) => {
...@@ -343,8 +373,57 @@ const cleanupFixture = Effect.gen(function* () {...@@ -343,8 +373,57 @@ const cleanupFixture = Effect.gen(function* () {
343 openSessionCwd = cwd;373 openSessionCwd = cwd;
344 },374 },
345 fullStatusCalls: () => fullStatusCalls,375 fullStatusCalls: () => fullStatusCalls,
376 snapshotReads: () => snapshotReads,
377 /** Starts the hourly sweep and the subscription to `domainEvents`, which wait until now. */
378 activate: (domainEvents: Stream.Stream<OrchestrationV2DomainEvent>) => {
379 events = domainEvents;
380 return Deferred.succeed(activated, undefined);
381 },
346 };382 };
347});383});
384/** The fixture, with the size walk of its worktree held until `resume` succeeds. */
385const pausedMeasurementFixture = Effect.gen(function* () {
386 const fs = yield* FileSystem.FileSystem;
387 const measuring = yield* Deferred.make<void>();
388 const resume = yield* Deferred.make<void>();
389 let measuredPath: string | undefined;
390 const fixture = yield* cleanupFixture.pipe(
391 Effect.provideService(FileSystem.FileSystem, {
392 ...fs,
393 readDirectory: (directory, options) =>
394 Effect.gen(function* () {
395 if (directory === measuredPath) {
396 yield* Deferred.succeed(measuring, undefined);
397 yield* Deferred.await(resume);
398 }
399 return yield* fs.readDirectory(directory, options);
400 }),
401 }),
402 );
403 measuredPath = yield* fs.realPath(fixture.worktree);
404 return { ...fixture, measuring, resume };
405});
406const sessionUpdated = (
407 index: number,
408 status: OrchestrationV2ProviderSession["status"],
409): OrchestrationV2DomainEvent => ({
410 id: EventId.make(`event-${index}`),
411 type: "provider-session.updated",
412 threadId: ThreadId.make("thread-1"),
413 occurredAt: at(0),
414 payload: {
415 id: ProviderSessionId.make(`session-${index}`),
416 driver: ProviderDriverKind.make("codex"),
417 providerInstanceId: ProviderInstanceId.make("codex"),
418 status,
419 cwd: "/elsewhere",
420 model: null,
421 capabilities: CodexProviderCapabilitiesV2,
422 createdAt: at(0),
423 updatedAt: at(0),
424 lastError: null,
425 },
426});
348const cleanupTestLayer = GitVcsDriver.layer.pipe(427const cleanupTestLayer = GitVcsDriver.layer.pipe(
349 Layer.provideMerge(ServerConfig.layerTest(process.cwd(), { prefix: "storage-cleanup-" })),428 Layer.provideMerge(ServerConfig.layerTest(process.cwd(), { prefix: "storage-cleanup-" })),
350 Layer.provideMerge(NodeServices.layer),429 Layer.provideMerge(NodeServices.layer),
...@@ -402,35 +481,31 @@ describe("storage cleanup reports and local file policies", () => {...@@ -402,35 +481,31 @@ describe("storage cleanup reports and local file policies", () => {
402 }),481 }),
403 ),482 ),
404 );483 );
405 it.live("keeps a worktree when its thread starts during size measurement", () =>484 it.live.each([
485 [
486 "its thread starts",
487 (worktree: string) => [
488 shell({ branch: "feature", worktreePath: worktree, status: "running" }),
489 ],
490 ],
491 [
492 "another thread starts using it",
493 (worktree: string) => [
494 shell({ branch: "feature", worktreePath: worktree }),
495 shell({ id: ThreadId.make("thread-2"), branch: "feature", worktreePath: worktree }),
496 ],
497 ],
498 ] as const)("keeps a worktree when %s during size measurement", ([, threadsWhileMeasuring]) =>
406 runCleanupTest(499 runCleanupTest(
407 Effect.gen(function* () {500 Effect.gen(function* () {
408 const fs = yield* FileSystem.FileSystem;501 const fixture = yield* pausedMeasurementFixture;
409 const measuring = yield* Deferred.make<void>();502 const { fs, worktree } = fixture;
410 const resume = yield* Deferred.make<void>();
411 let measuredPath: string | undefined;
412 const fixture = yield* cleanupFixture.pipe(
413 Effect.provideService(FileSystem.FileSystem, {
414 ...fs,
415 readDirectory: (directory) =>
416 Effect.gen(function* () {
417 if (directory === measuredPath) {
418 yield* Deferred.succeed(measuring, undefined);
419 yield* Deferred.await(resume);
420 }
421 return yield* fs.readDirectory(directory);
422 }),
423 }),
424 );
425 measuredPath = yield* fs.realPath(fixture.worktree);
426 fixture.setPolicy("tracked-changes");503 fixture.setPolicy("tracked-changes");
427 yield* fs.writeFileString(`${fixture.worktree}/notes.txt`, "untracked\n");504 yield* fs.writeFileString(`${worktree}/notes.txt`, "untracked\n");
428 const cleanup = yield* fixture.service.runNow.pipe(Effect.forkChild);505 const cleanup = yield* fixture.service.runNow.pipe(Effect.forkChild);
429 yield* Deferred.await(measuring);506 yield* Deferred.await(fixture.measuring);
430 fixture.setThreads([507 fixture.setThreads(threadsWhileMeasuring(worktree));
431 shell({ branch: "feature", worktreePath: fixture.worktree, status: "running" }),508 yield* Deferred.succeed(fixture.resume, undefined);
432 ]);
433 yield* Deferred.succeed(resume, undefined);
434 const report = yield* Fiber.join(cleanup);509 const report = yield* Fiber.join(cleanup);
435 expect(report.entries[0]).toMatchObject({510 expect(report.entries[0]).toMatchObject({
436 outcome: "kept",511 outcome: "kept",
...@@ -438,11 +513,48 @@ describe("storage cleanup reports and local file policies", () => {...@@ -438,11 +513,48 @@ describe("storage cleanup reports and local file policies", () => {
438 bytes: null,513 bytes: null,
439 });514 });
440 expect(report.bytesFreed).toBe(0);515 expect(report.bytesFreed).toBe(0);
441 expect(yield* fs.exists(fixture.worktree)).toBe(true);516 expect(yield* fs.exists(worktree)).toBe(true);
442 expect(yield* fs.readFileString(`${fixture.worktree}/notes.txt`)).toBe("untracked\n");517 expect(yield* fs.readFileString(`${worktree}/notes.txt`)).toBe("untracked\n");
443 }),518 }),
444 ),519 ),
445 );520 );
521 it.live("reads the thread snapshot once per sweep, not per worktree", () =>
522 runCleanupTest(
523 Effect.gen(function* () {
524 const { service, snapshotReads } = yield* cleanupFixture;
525 expect((yield* service.runNow).entries[0]).toMatchObject({ outcome: "removed" });
526 expect(snapshotReads()).toBe(1);
527 }),
528 ),
529 );
530 it.live.each([
531 ["running", "ready", 1],
532 ["stopped", "error", 2],
533 ] as const)(
534 "session updates to %s or %s during a sweep make %i sweep(s) in total",
535 ([first, second, sweeps]) =>
536 runCleanupTest(
537 Effect.gen(function* () {
538 const fixture = yield* pausedMeasurementFixture;
539 const handled = yield* Deferred.make<void>();
540 // 20 updates arrive while the first automatic sweep measures the worktree.
541 yield* fixture.activate(
542 Stream.fromEffect(Deferred.await(fixture.measuring)).pipe(
543 Stream.flatMap(() => Stream.range(0, 19)),
544 Stream.map((index) => sessionUpdated(index, index % 2 === 0 ? first : second)),
545 Stream.concat(
546 Stream.fromEffect(Deferred.succeed(handled, undefined)).pipe(Stream.drain),
547 ),
548 ),
549 );
550 yield* Deferred.await(handled);
551 yield* Deferred.succeed(fixture.resume, undefined);
552 yield* fixture.service.drain;
553 // One snapshot read per sweep.
554 expect(fixture.snapshotReads()).toBe(sweeps);
555 }),
556 ),
557 );
446 it.live("does not measure a worktree kept for an open provider session", () =>558 it.live("does not measure a worktree kept for an open provider session", () =>
447 runCleanupTest(559 runCleanupTest(
448 Effect.gen(function* () {560 Effect.gen(function* () {
apps/server/src/storageCleanup.ts+49-27
...@@ -27,6 +27,7 @@ import * as Equal from "effect/Equal";...@@ -27,6 +27,7 @@ import * as Equal from "effect/Equal";
27import * as FileSystem from "effect/FileSystem";27import * as FileSystem from "effect/FileSystem";
28import * as Option from "effect/Option";28import * as Option from "effect/Option";
29import * as Path from "effect/Path";29import * as Path from "effect/Path";
30import * as Ref from "effect/Ref";
30import type { PlatformError } from "effect/PlatformError";31import type { PlatformError } from "effect/PlatformError";
31import * as Schedule from "effect/Schedule";32import * as Schedule from "effect/Schedule";
32import * as Stream from "effect/Stream";33import * as Stream from "effect/Stream";
...@@ -45,6 +46,7 @@ import { isFilesystemRoot, managedWorktreesDirectories } from "./worktreesDirect...@@ -45,6 +46,7 @@ import { isFilesystemRoot, managedWorktreesDirectories } from "./worktreesDirect
45import * as GitVcsDriver from "./vcs/GitVcsDriver.ts";46import * as GitVcsDriver from "./vcs/GitVcsDriver.ts";
46import { withWorkspaceLease } from "./workspace/workspaceLease.ts";47import { withWorkspaceLease } from "./workspace/workspaceLease.ts";
47import * as HostProcess from "@t3tools/shared/HostProcess";48import * as HostProcess from "@t3tools/shared/HostProcess";
49import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
4850
49const decodeCleanupThread = Schema.decodeUnknownEffect(51const decodeCleanupThread = Schema.decodeUnknownEffect(
50 Schema.fromJsonString(OrchestrationV2AppThreadJson),52 Schema.fromJsonString(OrchestrationV2AppThreadJson),
...@@ -248,11 +250,16 @@ const make = Effect.gen(function* () {...@@ -248,11 +250,16 @@ const make = Effect.gen(function* () {
248 Effect.orElseSucceed(() => null),250 Effect.orElseSucceed(() => null),
249 );251 );
250252
253 // Active threads only: archived threads are not candidates and do not count
254 // as sharing a worktree. The recheck's getThreadWorktreePaths reads the same set.
255 // The snapshot takes seconds on large databases, so it runs on the reader
256 // worker instead of the event loop.
251 const readThreads = Effect.fn("StorageCleanup.readThreads")(function* () {257 const readThreads = Effect.fn("StorageCleanup.readThreads")(function* () {
252 const active = yield* projections.getShellSnapshot();258 const { threads } = yield* projections
253 const archived = yield* projections.getShellSnapshot({ location: "archive" });259 .getShellSnapshot({ location: "active" })
260 .pipe(NodeSqliteClient.readOnly);
254 const projects = yield* projectStore.listShells();261 const projects = yield* projectStore.listShells();
255 return { projects, threads: [...active.threads, ...archived.threads] };262 return { projects, threads };
256 });263 });
257264
258 // Local threads under another project need not have a worktreePath of their own.265 // Local threads under another project need not have a worktreePath of their own.
...@@ -538,14 +545,14 @@ const make = Effect.gen(function* () {...@@ -538,14 +545,14 @@ const make = Effect.gen(function* () {
538 );545 );
539 // Threads, queued deletion work and provider sessions as they are now, so a546 // Threads, queued deletion work and provider sessions as they are now, so a
540 // queued turn, resumed session or new thread sharing this path cancels547 // queued turn, resumed session or new thread sharing this path cancels
541 // the removal.548 // the removal. Read only this path and this thread: a full snapshot per
549 // worktree takes seconds on large databases.
542 const blockedSinceCheck = Effect.fnUntraced(function* () {550 const blockedSinceCheck = Effect.fnUntraced(function* () {
543 const latestSnapshot = yield* readThreads();551 const latestProjects = yield* projectStore.listShells();
544 if (yield* containsProjectRoot(worktreePath, [project, ...latestSnapshot.projects]))552 if (yield* containsProjectRoot(worktreePath, [project, ...latestProjects]))
545 return "contains a project checkout";553 return "contains a project checkout";
546 const latest = latestSnapshot.threads.filter(554 const latest = (yield* projections.getThreadWorktreePaths()).filter(
547 (entry) =>555 (entry) => path.resolve(entry.worktreePath) === worktreePath,
548 entry.worktreePath !== null && path.resolve(entry.worktreePath) === worktreePath,
549 );556 );
550 if (hasTerminal(worktreePath)) return "open terminal";557 if (hasTerminal(worktreePath)) return "open terminal";
551 if (deleted) {558 if (deleted) {
...@@ -562,13 +569,18 @@ const make = Effect.gen(function* () {...@@ -562,13 +569,18 @@ const make = Effect.gen(function* () {
562 WHERE thread_id = ${thread.id} AND status NOT IN ('succeeded', 'cancelled') LIMIT 1569 WHERE thread_id = ${thread.id} AND status NOT IN ('succeeded', 'cancelled') LIMIT 1
563 `;570 `;
564 if (pendingCleanup.length > 0) return "thread deletion is still pending";571 if (pendingCleanup.length > 0) return "thread deletion is still pending";
565 } else if (572 } else {
566 latest.length !== 1 ||573 const latestShell =
567 latest[0]!.id !== thread.id ||574 latest.length === 1 && latest[0]!.threadId === thread.id
568 !storageCleanupThreadIdle(latest[0]!, now) ||575 ? yield* projections.getThreadShell(thread.id)
569 storageCleanupActivityAt(latest[0]!) !== storageCleanupActivityAt(thread)576 : null;
570 )577 if (
571 return "thread activity or shared worktree changed since check";578 latestShell === null ||
579 !storageCleanupThreadIdle(latestShell, now) ||
580 storageCleanupActivityAt(latestShell) !== storageCleanupActivityAt(thread)
581 )
582 return "thread activity or shared worktree changed since check";
583 }
572 // Sessions can outlive their run and can be shared across app threads.584 // Sessions can outlive their run and can be shared across app threads.
573 const sessionRows = yield* sql<{ payload_json: string }>`585 const sessionRows = yield* sql<{ payload_json: string }>`
574 SELECT payload_json FROM orchestration_v2_projection_provider_sessions586 SELECT payload_json FROM orchestration_v2_projection_provider_sessions
...@@ -775,9 +787,14 @@ const make = Effect.gen(function* () {...@@ -775,9 +787,14 @@ const make = Effect.gen(function* () {
775 yield* SubscriptionRef.set(reportRef, latestReport);787 yield* SubscriptionRef.set(reportRef, latestReport);
776 return latestReport;788 return latestReport;
777 });789 });
790 // At most one automatic sweep waits at a time. An automatic request made
791 // while one is waiting joins it, since that sweep has not read any state yet.
792 // Each manual run still gets its own sweep and report.
793 const automaticSweepQueued = yield* Ref.make(false);
778 const worker = yield* makeDrainableWorker(794 const worker = yield* makeDrainableWorker(
779 (completion: Deferred.Deferred<StorageCleanupReport, ServerSettingsError> | undefined) =>795 (completion: Deferred.Deferred<StorageCleanupReport, ServerSettingsError> | undefined) =>
780 sweep(completion === undefined ? "automatic" : "manual").pipe(796 (completion === undefined ? Ref.set(automaticSweepQueued, false) : Effect.void).pipe(
797 Effect.andThen(sweep(completion === undefined ? "automatic" : "manual")),
781 Effect.exit,798 Effect.exit,
782 Effect.flatMap((exit) =>799 Effect.flatMap((exit) =>
783 completion === undefined800 completion === undefined
...@@ -791,6 +808,9 @@ const make = Effect.gen(function* () {...@@ -791,6 +808,9 @@ const make = Effect.gen(function* () {
791 ),808 ),
792 ),809 ),
793 );810 );
811 const requestSweep = Ref.getAndSet(automaticSweepQueued, true).pipe(
812 Effect.flatMap((queued) => (queued ? Effect.void : worker.enqueue(undefined))),
813 );
794 const runNow = Effect.gen(function* () {814 const runNow = Effect.gen(function* () {
795 const completion = yield* Deferred.make<StorageCleanupReport, ServerSettingsError>();815 const completion = yield* Deferred.make<StorageCleanupReport, ServerSettingsError>();
796 yield* worker.enqueue(completion);816 yield* worker.enqueue(completion);
...@@ -817,13 +837,11 @@ const make = Effect.gen(function* () {...@@ -817,13 +837,11 @@ const make = Effect.gen(function* () {
817 const events = engine.streamDomainEvents;837 const events = engine.streamDomainEvents;
818 let lastSettings = yield* settingsService.getSettings.pipe(Effect.orDie);838 let lastSettings = yield* settingsService.getSettings.pipe(Effect.orDie);
819 yield* forkParked(839 yield* forkParked(
820 worker840 requestSweep.pipe(
821 .enqueue(undefined)841 Effect.andThen(worker.drain),
822 .pipe(842 Effect.repeat(Schedule.spaced("1 hour")),
823 Effect.andThen(worker.drain),843 Effect.asVoid,
824 Effect.repeat(Schedule.spaced("1 hour")),844 ),
825 Effect.asVoid,
826 ),
827 );845 );
828 yield* forkParked(846 yield* forkParked(
829 Stream.runForEach(changes, (settings) => {847 Stream.runForEach(changes, (settings) => {
...@@ -834,14 +852,18 @@ const make = Effect.gen(function* () {...@@ -834,14 +852,18 @@ const make = Effect.gen(function* () {
834 )852 )
835 return Effect.void;853 return Effect.void;
836 lastSettings = settings;854 lastSettings = settings;
837 return worker.enqueue(undefined);855 return requestSweep;
838 }),856 }),
839 );857 );
840 yield* forkParked(858 yield* forkParked(
859 // A deleted thread's worktree waits for its sessions to end. Moving
860 // between live statuses, such as running to ready, cannot free one.
841 Stream.runForEach(events, (event) =>861 Stream.runForEach(events, (event) =>
842 (event.type === "thread.deleted" || event.type === "provider-session.updated") &&862 (event.type === "thread.deleted" ||
863 (event.type === "provider-session.updated" &&
864 (event.payload.status === "stopped" || event.payload.status === "error"))) &&
843 anyWorktreePolicy(lastSettings, (rules) => rules.worktreeOnDelete)865 anyWorktreePolicy(lastSettings, (rules) => rules.worktreeOnDelete)
844 ? worker.enqueue(undefined)866 ? requestSweep
845 : Effect.void,867 : Effect.void,
846 ).pipe(868 ).pipe(
847 Effect.catchCause((cause) =>869 Effect.catchCause((cause) =>