diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index cda6f9ab3ac..75bcb225075 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -2596,6 +2596,11 @@ PR merge or delivery status. Enable it with reclaimed bytes, base removal, and skip reasons. The existing bounded lifecycle sweep supplies scheduling and shutdown cancellation. +The eligibility query first rejects paths shared with another execution or +project checkout. This prevents a known shared-path veto from scanning retained +run contexts. Unshared paths still require every existing task, run, recovery, +lease, service and filesystem check in the same lifecycle lock. + Before deletion the worker fetches the exact commit into a fresh object store using the company's normal Git identity and commits a versioned restoration recipe to workspace metadata. Current runs, previews, pending reopening, diff --git a/server/src/__tests__/git-workspace-expiration-eligibility.test.ts b/server/src/__tests__/git-workspace-expiration-eligibility.test.ts new file mode 100644 index 00000000000..a577f4eee4e --- /dev/null +++ b/server/src/__tests__/git-workspace-expiration-eligibility.test.ts @@ -0,0 +1,94 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { eq, sql } from "drizzle-orm"; +import { agents, companies, createDb, executionWorkspaces, heartbeatRuns, issues, projects, projectWorkspaces } from "@paperclipai/db"; +import { isGitWorkspaceExpirationEligible } from "../services/git-workspace-expiration.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; + +type Plan = { "Relation Name"?: string; "Actual Loops"?: number; Plans?: Plan[] }; +const flatten = (plan: Plan): Plan[] => [plan, ...(plan.Plans ?? []).flatMap(flatten)]; +const support = await getEmbeddedPostgresTestSupport(); +const describeDatabase = support.supported ? describe : describe.skip; + +describeDatabase("Git expiration eligibility", () => { + let database: Awaited>; + let db: ReturnType; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("git-expiration-eligibility-"); + db = createDb(database.connectionString, { maxConnections: 1 }); + }); + afterEach(async () => { vi.restoreAllMocks(); await db.execute(sql`TRUNCATE TABLE companies CASCADE`); }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await database?.cleanup(); }); + + async function fixture() { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(); + const now = new Date(), old = new Date(now.getTime() - 3 * 86_400_000); + await db.insert(companies).values({ id: companyId, name: "Eligibility", issuePrefix: `EL${companyId.slice(0, 6)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Eligibility" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Finished root", status: "done", completedAt: old }); + const [workspace] = await db.insert(executionWorkspaces).values({ companyId, sourceIssueId: issueId, + mode: "isolated_workspace", strategyType: "git_worktree", name: "Eligibility", cwd: `/tmp/eligibility-${companyId}` }).returning(); + await db.insert(heartbeatRuns).values({ companyId, agentId, status: "succeeded", finishedAt: old, contextSnapshot: { issueId } }); + const eligible = () => db.transaction(tx => isGitWorkspaceExpirationEligible(tx, workspace, now)); + return { companyId, agentId, issueId, workspace, now, old, eligible }; + } + + async function alias(f: Awaited>, kind: string) { + const companyId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Other owner", issuePrefix: `EO${companyId.slice(0, 6)}` }); + const cwd = kind.endsWith("child") ? `${f.workspace.cwd}/nested` : f.workspace.cwd; + if (kind.startsWith("project")) { + const [project] = await db.insert(projects).values({ companyId, name: "Other project" }).returning(); + const [row] = await db.insert(projectWorkspaces).values({ companyId, projectId: project.id, name: "Other checkout", cwd }).returning(); + return { id: row.id, table: projectWorkspaces }; + } + const [row] = await db.insert(executionWorkspaces).values({ companyId, mode: "isolated_workspace", strategyType: "git_worktree", name: "Other checkout", + cwd: kind === "workspace provider" ? `/tmp/unrelated-${companyId}` : cwd, + providerRef: kind === "workspace provider" ? f.workspace.cwd : null }).returning(); + return { id: row.id, table: executionWorkspaces }; + } + + it.each(["workspace same", "workspace provider", "workspace child", "project same", "project child"])("rejects %s before reading run history", async kind => { + const f = await fixture(); + await alias(f, kind); + await db.transaction(async tx => { + const execute = vi.spyOn(tx, "execute"); + expect(await isGitWorkspaceExpirationEligible(tx, f.workspace, f.now)).toBe(false); + const query = execute.mock.calls[0]![0]; + execute.mockRestore(); + const rows = await tx.execute(sql`EXPLAIN (ANALYZE, FORMAT JSON) ${query}`); + const explained = rows[0]["QUERY PLAN"] as Array<{ Plan: Plan }>; + const historyScans = flatten(explained[0].Plan).filter(node => node["Relation Name"] === "heartbeat_runs"); + expect(historyScans.every(node => node["Actual Loops"] === 0)).toBe(true); + }); + }); + + it("rechecks a shared consumer added or removed between eligibility checks", async () => { + const f = await fixture(); + expect(await f.eligible()).toBe(true); + const consumer = await alias(f, "workspace same"); + expect(await f.eligible()).toBe(false); + await db.delete(consumer.table).where(eq(consumer.table.id, consumer.id)); + expect(await f.eligible()).toBe(true); + }); + + it("does not treat a sibling path prefix as a descendant", async () => { + const f = await fixture(); + await db.insert(executionWorkspaces).values({ companyId: f.companyId, mode: "isolated_workspace", strategyType: "git_worktree", name: "Sibling", cwd: `${f.workspace.cwd}-sibling` }); + expect(await f.eligible()).toBe(true); + }); + + it.each(["issueId", "nativeIssueId", "contextIssue", "contextTask", "contextWorkspace", "checkout"])("retains the active-run veto through %s", async route => { + const f = await fixture(), runId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: runId, companyId: f.companyId, agentId: f.agentId, status: "scheduled_retry", + ...(route === "issueId" ? { scopeKind: "issue" as const, issueId: f.issueId } : {}), + ...(route === "nativeIssueId" ? { nativeIssueId: f.issueId } : {}), + contextSnapshot: route === "contextIssue" ? { issueId: f.issueId } : route === "contextTask" ? { taskId: f.issueId } : route === "contextWorkspace" ? { executionWorkspaceId: f.workspace.id } : null }); + if (route === "checkout") { + await db.insert(issues).values({ companyId: f.companyId, title: "Other finished consumer", status: "done", completedAt: f.old, executionWorkspaceId: f.workspace.id, checkoutRunId: runId }); + } + expect(await f.eligible()).toBe(false); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: f.old }).where(eq(heartbeatRuns.id, runId)); + expect(await f.eligible()).toBe(true); + }); +}); diff --git a/server/src/services/git-workspace-expiration.ts b/server/src/services/git-workspace-expiration.ts index 4a5c56edbef..726f5ea24e4 100644 --- a/server/src/services/git-workspace-expiration.ts +++ b/server/src/services/git-workspace-expiration.ts @@ -32,7 +32,7 @@ export type GitExpirationResult = { candidate: boolean; expired: boolean; reclai candidateBytes: number; baseExpired: boolean; skipReason: string | null }; /** Re-read the whole tree and every runtime/recovery consumer immediately before deletion. */ -async function eligible(tx: Tx, workspace: Workspace, now: Date) { +export async function isGitWorkspaceExpirationEligible(tx: Tx, workspace: Workspace, now: Date) { if (workspace.metadata?.reopenPendingConsumption === true) return false; const rows = await tx.execute<{ allowed: boolean }>(sql` WITH RECURSIVE tree AS ( @@ -50,7 +50,13 @@ async function eligible(tx: Tx, workspace: Workspace, now: Date) { OR r.id IN (SELECT unnest(ARRAY[checkout_run_id, execution_run_id]) FROM ${issues} WHERE company_id = ${workspace.companyId} AND execution_workspace_id = ${workspace.id}) ) - ) SELECT ( + ) SELECT CASE WHEN + EXISTS (SELECT 1 FROM ${executionWorkspaces} other WHERE other.id <> ${workspace.id} + AND (other.cwd = ${workspace.cwd} OR other.provider_ref = ${workspace.cwd} + OR left(other.cwd, length(${workspace.cwd}) + 1) = ${workspace.cwd} || '/')) + OR EXISTS (SELECT 1 FROM ${projectWorkspaces} WHERE cwd = ${workspace.cwd} + OR left(cwd, length(${workspace.cwd}) + 1) = ${workspace.cwd} || '/') + THEN false ELSE ( EXISTS (SELECT 1 FROM tree) AND NOT EXISTS (SELECT 1 FROM tree WHERE status NOT IN ('done', 'cancelled') OR coalesce(CASE WHEN status = 'done' THEN completed_at ELSE cancelled_at END, updated_at) > ${new Date(now.getTime() - DAY_MS).toISOString()}::timestamptz) @@ -70,12 +76,8 @@ async function eligible(tx: Tx, workspace: Workspace, now: Date) { AND NOT EXISTS (SELECT 1 FROM ${workspaceRuntimeServices} WHERE company_id = ${workspace.companyId} AND (execution_workspace_id = ${workspace.id} OR cwd = ${workspace.cwd} OR left(cwd, length(${workspace.cwd}) + 1) = ${workspace.cwd} || '/') AND status <> 'stopped') - AND NOT EXISTS (SELECT 1 FROM ${executionWorkspaces} other WHERE other.id <> ${workspace.id} - AND (other.cwd = ${workspace.cwd} OR other.provider_ref = ${workspace.cwd} - OR left(other.cwd, length(${workspace.cwd}) + 1) = ${workspace.cwd} || '/')) - AND NOT EXISTS (SELECT 1 FROM ${projectWorkspaces} WHERE cwd = ${workspace.cwd} - OR left(cwd, length(${workspace.cwd}) + 1) = ${workspace.cwd} || '/') - ) AS allowed`); + + ) END AS allowed`); return rows[0]?.allowed === true; } @@ -158,7 +160,7 @@ export async function expireReconstructableGitWorkspace(db: Db, snapshot: Worksp if (!acquired[0]?.locked) throw new Error("lifecycle_busy"); let workspace = (await db.select().from(executionWorkspaces).where(and(eq(executionWorkspaces.id, snapshot.id), eq(executionWorkspaces.companyId, snapshot.companyId))))[0]; if (!workspace || workspace.updatedAt.getTime() !== snapshot.updatedAt.getTime()) throw new Error("workspace_changed"); - if (!await eligible(tx, workspace, options.now)) throw new Error("live_or_unresolved_consumer"); + if (!await isGitWorkspaceExpirationEligible(tx, workspace, options.now)) throw new Error("live_or_unresolved_consumer"); let state = ["active", "idle", "in_review"].includes(workspace.status) ? null : readGitCacheExpiration(workspace); // A restored active workspace starts a new expiration cycle and records its new HEAD. if (!hasPendingGitExpiration(workspace) && !["active", "idle", "in_review"].includes(workspace.status)) throw new Error("not_expirable"); @@ -204,7 +206,7 @@ export async function expireReconstructableGitWorkspace(db: Db, snapshot: Worksp .where(eq(executionWorkspaces.id, snapshot.id)).returning(); }; await options.beforeDelete?.(workspace); - if (!await eligible(tx, workspace, options.now)) throw new Error("live_or_unresolved_consumer"); + if (!await isGitWorkspaceExpirationEligible(tx, workspace, options.now)) throw new Error("live_or_unresolved_consumer"); if (!state.workspaceRemoved && present) { const { acquireGitWorktreeCleanupLock } = await import("./workspace-runtime.js"); const lock = await acquireGitWorktreeCleanupLock(recipe.worktreePath); @@ -224,7 +226,7 @@ export async function expireReconstructableGitWorkspace(db: Db, snapshot: Worksp try { if ((!state.baseTrash || !await exists(state.baseTrash)) && await exists(recipe.sourcePath)) { const bytes = await inspectBase(db, workspace!, options.signal); - if (!await eligible(tx, workspace!, options.now)) throw new Error("live_or_unresolved_consumer"); + if (!await isGitWorkspaceExpirationEligible(tx, workspace!, options.now)) throw new Error("live_or_unresolved_consumer"); const trash = path.join(path.dirname(recipe.sourcePath), `.expired-git-base-${snapshot.id}-${state.generation}`); if (await exists(trash)) throw new Error("unknown_base_trash"); state.baseTrash = trash;