Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
112 changes: 106 additions & 6 deletions apps/server/src/storageCleanup.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -262,6 +262,7 @@ const cleanupFixture = (
storageCleanup,
};
let threads = [shell({ branch: "feature", worktreePath: worktree })];
let deletedThreads: OrchestrationV2ThreadShell[] = [];
let storedWorktreePaths: string[] = [worktree];
let projectRoots = [repo];
let editBeforeRemoval = false;
Expand Down Expand Up @@ -358,12 +359,17 @@ const cleanupFixture = (
}),
},
]
: strings.join("").includes('AS "worktreePath"')
? [
...threads.flatMap((thread) => thread.worktreePath ?? []),
...storedWorktreePaths,
].map((worktreePath) => ({ worktreePath }))
: [],
: strings.join("").includes("deleted_at IS NOT NULL")
? deletedThreads.map((thread) => ({
payload_json: JSON.stringify(thread),
workspaceRoot: repo,
}))
: strings.join("").includes('AS "worktreePath"')
? [
...threads.flatMap((thread) => thread.worktreePath ?? []),
...storedWorktreePaths,
].map((worktreePath) => ({ worktreePath }))
: [],
)) as unknown as SqlClient.SqlClient),
Effect.provideService(GitManager.GitManager, {
invalidateStatus: () => Effect.void,
Expand Down Expand Up @@ -415,6 +421,9 @@ const cleanupFixture = (
setThreads: (next: typeof threads) => {
threads = next;
},
setDeletedThreads: (next: OrchestrationV2ThreadShell[]) => {
deletedThreads = next;
},
setStoredWorktreePaths: (paths: string[]) => {
storedWorktreePaths = paths;
},
Expand Down Expand Up @@ -848,6 +857,97 @@ describe("subagent worktree ownership", () => {
);
});

describe("deleted thread worktrees", () => {
const parent = (worktreePath: string) =>
shell({ branch: "feature", worktreePath, deletedAt: at(-DAY_MS) });
// Deleting the parent leaves its native subagent thread in the projection.
const subagent = (
owner: OrchestrationV2ThreadShell,
overrides: Partial<OrchestrationV2ThreadShell> = {},
) =>
shell({
id: ThreadId.make("child"),
title: "Subagent",
branch: "feature",
worktreePath: owner.worktreePath,
status: "completed",
createdBy: "agent",
creationSource: "provider",
lineage: {
rootThreadId: owner.id,
parentThreadId: owner.id,
relationshipToParent: "subagent",
},
...overrides,
});

it.effect.each(["no subagents", "native subagent"] as const)(
"removes a deleted thread's worktree with %s",
(scenario) =>
Effect.gen(function* () {
yield* TestClock.setTime(NOW_MS);
const fixture = yield* cleanupFixture({
...DEFAULT_SERVER_SETTINGS.storageCleanup,
worktreeAfterDays: null,
worktreeOnDelete: true,
});
const owner = parent(fixture.worktree);
fixture.setDeletedThreads([owner]);
fixture.setThreads(scenario === "native subagent" ? [subagent(owner)] : []);
const report = yield* fixture.service.runNow;
expect(report.entries[0]).toMatchObject({
threadId: owner.id,
outcome: "removed",
reason: "Thread was deleted",
});
expect(yield* fixture.fs.exists(fixture.worktree)).toBe(false);
}).pipe(Effect.provide(cleanupTestLayer), Effect.scoped),
);

it.effect("applies the remaining subagent's rules when deletion cleanup is off", () =>
Effect.gen(function* () {
yield* TestClock.setTime(NOW_MS);
const fixture = yield* cleanupFixture({
...DEFAULT_SERVER_SETTINGS.storageCleanup,
worktreeAfterDays: 8,
worktreeOnDelete: false,
});
const owner = parent(fixture.worktree);
const child = subagent(owner);
fixture.setDeletedThreads([owner]);
fixture.setThreads([child]);
const report = yield* fixture.service.runNow;
expect(report.entries[0]).toMatchObject({
threadId: child.id,
outcome: "removed",
reason: "Inactive for 30 days",
});
expect(yield* fixture.fs.exists(fixture.worktree)).toBe(false);
}).pipe(Effect.provide(cleanupTestLayer), Effect.scoped),
);

it.effect("waits out the settle grace of a subagent that just settled", () =>
Effect.gen(function* () {
yield* TestClock.setTime(NOW_MS);
const fixture = yield* cleanupFixture({
...DEFAULT_SERVER_SETTINGS.storageCleanup,
worktreeAfterDays: null,
worktreeOnDelete: true,
});
const owner = parent(fixture.worktree);
fixture.setDeletedThreads([owner]);
fixture.setThreads([subagent(owner, { settledAt: at(0), updatedAt: at(0) })]);
const report = yield* fixture.service.runNow;
expect(report.entries[0]).toMatchObject({
threadId: owner.id,
outcome: "kept",
reason: "Settled moments ago",
});
expect(yield* fixture.fs.exists(fixture.worktree)).toBe(true);
}).pipe(Effect.provide(cleanupTestLayer), Effect.scoped),
);
});

describe("orphan worktree reports", () => {
it.live("keeps orphan results visible when thread results fill the report", () =>
runCleanupTest(
Expand Down
79 changes: 47 additions & 32 deletions apps/server/src/storageCleanup.ts
Original file line number Diff line number Diff line change
Expand Up @@ -146,7 +146,9 @@ export function storageCleanupActivityAt(thread: OrchestrationV2ThreadShell): nu
}

/** Only a complete subagent chain within this worktree shares ownership. */
function worktreeOwner(threads: ReadonlyArray<OrchestrationV2ThreadShell>) {
function worktreeOwner<T extends Pick<OrchestrationV2ThreadShell, "id" | "lineage">>(
threads: ReadonlyArray<T>,
) {
if (threads.length === 1) return threads[0];
const byId = new Map(threads.map((thread) => [thread.id, thread]));
const tops = threads.filter((thread) => !byId.has(thread.lineage.parentThreadId!));
Expand Down Expand Up @@ -750,8 +752,24 @@ const make = Effect.gen(function* () {
snapshot.threads.filter((thread) => thread.worktreePath !== null),
(thread) => path.resolve(thread.worktreePath!),
);
const deletedByPath = Map.groupBy(deletedThreads, (thread) =>
path.resolve(thread.worktreePath!),
);
const candidates = [
...[...groups.values()].map((group) => worktreeOwner(group) ?? group[0]!),
...[...groups].map(
([worktreePath, group]) =>
// A deleted thread still owns the worktree its remaining subagents share,
// but only deletion cleanup acts on it; otherwise the live rules apply.
deletedByPath
.get(worktreePath)
?.find(
(deleted) =>
resolveWorktreeCleanup(serverSettings, deleted.projectId).worktreeOnDelete &&
worktreeOwner([deleted, ...group]) === deleted,
) ??
worktreeOwner(group) ??
group[0]!,
),
...deletedThreads.filter((thread) => !groups.has(path.resolve(thread.worktreePath!))),
];
for (const thread of candidates) {
Expand All @@ -761,6 +779,7 @@ const make = Effect.gen(function* () {
const group = groups.get(worktreePath) ?? [];
const activityAt = Math.max(...group.map(storageCleanupActivityAt));
const deleted = "workspaceRoot" in thread;
const members = deleted ? [thread, ...group] : group;
const project = deleted
? { workspaceRoot: thread.workspaceRoot }
: snapshot.projects.find((entry) => entry.id === thread.projectId);
Expand All @@ -779,12 +798,12 @@ const make = Effect.gen(function* () {
yield* Effect.gen(function* () {
if (!(yield* fs.exists(worktreePath))) return;
if (deleted && !settings.worktreeOnDelete) return keep("no rules apply");
if (group.length > 1 && worktreeOwner(group) === undefined)
return keep(`shared by ${group.length} threads`);
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 ||
(!deleted && !group.every((member) => storageCleanupThreadIdle(member, now)))
!group.every((member) => storageCleanupThreadIdle(member, now))
)
return keep("thread is running or has pending work");
if (group.some((member) => member.branch !== thread.branch))
Expand All @@ -807,7 +826,6 @@ const make = Effect.gen(function* () {
const changes = yield* localChangesReason(worktreePath, settings);
if (changes !== null) return keep(changes);
if (
!deleted &&
group.some(
(member) =>
member.settledAt !== null &&
Expand Down Expand Up @@ -928,7 +946,6 @@ const make = Effect.gen(function* () {
if (hasTerminal(worktreePath)) return "open terminal";
if (deleted) {
if (
latest.length > 0 ||
!resolveWorktreeCleanup(yield* settingsService.getSettings, thread.projectId)
.worktreeOnDelete
)
Expand All @@ -940,32 +957,30 @@ 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 {
const latestShells = yield* Effect.forEach(latest, (entry) =>
projections.getThreadShell(entry.threadId),
);
const latestGroup = latestShells.filter((shell) => shell !== null);
const latestOwner = worktreeOwner(latestGroup);
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 ||
!Equal.equals(shell.settledAt, previous.settledAt) ||
(shell.settledAt !== null &&
!Equal.equals(shell.updatedAt, previous.updatedAt)) ||
!storageCleanupThreadIdle(shell, now) ||
path.resolve(shell.worktreePath!) !== worktreePath ||
storageCleanupActivityAt(shell) !== storageCleanupActivityAt(previous)
);
})
)
return "thread activity or shared worktree changed since check";
}
const latestShells = yield* Effect.forEach(latest, (entry) =>
projections.getThreadShell(entry.threadId),
);
const latestGroup = latestShells.filter((shell) => shell !== null);
const latestOwner = worktreeOwner(deleted ? [thread, ...latestGroup] : latestGroup);
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 ||
!Equal.equals(shell.settledAt, previous.settledAt) ||
(shell.settledAt !== null && !Equal.equals(shell.updatedAt, previous.updatedAt)) ||
!storageCleanupThreadIdle(shell, now) ||
path.resolve(shell.worktreePath!) !== worktreePath ||
storageCleanupActivityAt(shell) !== storageCleanupActivityAt(previous)
);
})
)
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
Expand Down
Loading