diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 50aba9a3119..ecfb81fb92b 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -2603,6 +2603,9 @@ Environment overrides: `PAPERCLIP_GIT_RESTORE_POINT_TERMINAL_DELAY_DAYS=` sets how long the issue tree must be closed first. The default is `3`. A negative, empty, or non-numeric value falls back to the default. + `PAPERCLIP_GIT_RESTORE_POINT_IDLE_DAYS=` sets how long an open issue tree's + worktree must go untouched before it is removed the same way; runs restore it + automatically. The default is `7`, and `0` turns this off. - `PAPERCLIP_GIT_RESTORE_POINT_RETENTION_DAYS=` sets how long a git workspace restore point is kept after its issue tree closes (see `doc/recoverable-disk-cleanup.md`). The default is `90` and the minimum is `1`. A diff --git a/doc/recoverable-disk-cleanup.md b/doc/recoverable-disk-cleanup.md index 24955ea34d4..e3ea33ac73d 100644 --- a/doc/recoverable-disk-cleanup.md +++ b/doc/recoverable-disk-cleanup.md @@ -100,6 +100,15 @@ closed issue's worktree only when its remote can reproduce it exactly. With `PAPERCLIP_GIT_RESTORE_POINTS_ENABLED=true`, it can also remove a worktree that has unpushed commits, uncommitted or untracked files, or history-only commits, once its issue tree has been closed for `PAPERCLIP_GIT_RESTORE_POINT_TERMINAL_DELAY_DAYS` (default `3`). +It also removes the worktree of an issue tree that is still open once nothing has +touched it for `PAPERCLIP_GIT_RESTORE_POINT_IDLE_DAYS` (default `7`, `0` turns this off): +no issue in the tree or bound to the workspace was updated and no run for those issues or +the workspace was created, started or finished in that time, and no run is still live. An +interrupted removal is finished by later sweeps while the issue stays open. The workspace reaper sets a +deadline for that moment, so open trees cost no recurring checks before then. When a run +later finds its issue's workspace removed this way, it reopens the workspace, restore +point included, before using it; if the restore fails, the run fails instead of starting +in a new workspace. A comment or an agent update on the issue reopens it the same way. Before removal it saves a restore point: - a git bundle of commits the remote does not have, against a remote-verified base, diff --git a/server/src/__tests__/git-restore-point-expiration-config.test.ts b/server/src/__tests__/git-restore-point-expiration-config.test.ts index 5d1aacc3999..9e1fe3dd777 100644 --- a/server/src/__tests__/git-restore-point-expiration-config.test.ts +++ b/server/src/__tests__/git-restore-point-expiration-config.test.ts @@ -18,6 +18,13 @@ describe("git restore point expiration config parsing", () => { expect(loadConfig()).toMatchObject({ gitRestorePointsEnabled: true, gitRestorePointTerminalDelayDays: 0 }); }); + it("expires idle open issues after seven days by default, and never when set to zero", () => { + vi.stubEnv("PAPERCLIP_GIT_RESTORE_POINT_IDLE_DAYS", undefined); + expect(loadConfig()).toMatchObject({ gitRestorePointIdleDays: 7 }); + vi.stubEnv("PAPERCLIP_GIT_RESTORE_POINT_IDLE_DAYS", "0"); + expect(loadConfig()).toMatchObject({ gitRestorePointIdleDays: 0 }); + }); + it.each(["", "-1", "many"])("falls back to three days for %j", (value) => { vi.stubEnv("PAPERCLIP_GIT_RESTORE_POINT_TERMINAL_DELAY_DAYS", value); expect(loadConfig()).toMatchObject({ gitRestorePointTerminalDelayDays: 3 }); diff --git a/server/src/__tests__/git-restore-point-expiration.test.ts b/server/src/__tests__/git-restore-point-expiration.test.ts index fd27b0a24ae..b21b4b75745 100644 --- a/server/src/__tests__/git-restore-point-expiration.test.ts +++ b/server/src/__tests__/git-restore-point-expiration.test.ts @@ -6,12 +6,12 @@ import { randomUUID } from "node:crypto"; import { pathToFileURL } from "node:url"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { eq, sql } from "drizzle-orm"; -import { companies, createDb, executionWorkspaceRestorePoints, executionWorkspaces, issues } from "@paperclipai/db"; +import { agents, companies, createDb, executionWorkspaceRestorePoints, executionWorkspaces, heartbeatRuns, issues } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { materializeIsolatedRemoteWorkspaceSource } from "../services/isolated-remote-workspace-source.js"; import { ensurePersistedExecutionWorkspaceAvailable, realizeExecutionWorkspace } from "../services/workspace-runtime.js"; import { executionWorkspaceService } from "../services/execution-workspaces.js"; -import { expireReconstructableGitWorkspace } from "../services/git-workspace-expiration.js"; +import { expireReconstructableGitWorkspace, gitWorkspaceLastActivityAt } from "../services/git-workspace-expiration.js"; import { getStorageService } from "../storage/index.js"; const gitEnv = { ...process.env, GIT_AUTHOR_NAME: "Fixture", GIT_AUTHOR_EMAIL: "fixture@example.test", @@ -226,6 +226,104 @@ describeEmbeddedPostgres("git restore point expiration", () => { expect(await f.reopen()).toMatchObject({ ok: false, code: "rebuild_failed" }); }); + describe("idle open issues", () => { + async function openIdle(f: Awaited>, idleDays: number) { + await db.update(issues).set({ status: "in_progress", completedAt: null, updatedAt: new Date(Date.now() - idleDays * DAY_MS) }) + .where(eq(issues.id, f.issueId)); + } + const idle = (now = Date.now()) => ({ idleCutoff: new Date(now - 7 * DAY_MS) }); + + it("expires an open issue's idle worktree with a restore point and restores it on reopen", async () => { + const f = await fixture(); + const unpushed = await makeUnsavedWork(f.cwd); + await openIdle(f, 8); + expect(await f.expire()).toMatchObject({ expired: false, skipReason: "live_or_unresolved_consumer" }); + expect(await f.expire(idle())).toMatchObject({ expired: true, restorePoint: true, skipReason: null }); + expect(await exists(f.cwd)).toBe(false); + expect((await f.row()).status).toBe("archived"); + expect(await f.reopen()).toMatchObject({ ok: true, reopened: true }); + expect(git(f.cwd, "rev-parse", "HEAD")).toBe(unpushed); + expect(await fs.readFile(path.join(f.cwd, "notes", "todo.md"), "utf8")).toBe("untracked work\n"); + }); + + it("lets the workspace reaper expire idle open issues at their idle deadline", async () => { + const f = await fixture(); + await makeUnsavedWork(f.cwd); + const updatedAt = new Date(Date.now() - 3 * DAY_MS); + await db.update(issues).set({ status: "in_progress", completedAt: null, updatedAt }).where(eq(issues.id, f.issueId)); + const worker = (now: number) => executionWorkspaceService(db, { gitCacheExpirationEnabled: true, now: () => new Date(now), + gitRestorePoints: { storage: getStorageService(), terminalDelayMs: 3 * DAY_MS, keyDurable: true, idleMs: 7 * DAY_MS } }); + const early = worker(Date.now()); + expect(await early.sweepTerminalWorkspaces()).toMatchObject({ archived: 0, skippedNonTerminalTree: 1 }); + expect(await early.nextTerminalWorkspaceSweepAt()).toBe(updatedAt.getTime() + 7 * DAY_MS); + expect(await exists(f.cwd)).toBe(true); + const due = worker(updatedAt.getTime() + 7 * DAY_MS + 60_000); + expect(await due.sweepTerminalWorkspaces()).toMatchObject({ archived: 1 }); + expect(await exists(f.cwd)).toBe(false); + expect(await f.ledger()).toMatchObject([{ status: "active" }]); + }); + + it("lets the workspace reaper finish an interrupted idle expiration while the issue stays open", async () => { + const f = await fixture(); + await makeUnsavedWork(f.cwd); + await openIdle(f, 8); + expect(await f.expire({ ...idle(), beforeDelete: async () => { throw new Error("interrupted_fixture"); } })) + .toMatchObject({ expired: false }); + expect(await f.row()).toMatchObject({ status: "archived", metadata: { gitCacheExpiration: { completed: false } } }); + expect(await exists(f.cwd)).toBe(true); + const worker = executionWorkspaceService(db, { gitCacheExpirationEnabled: true, + gitRestorePoints: { storage: getStorageService(), terminalDelayMs: 3 * DAY_MS, keyDurable: true, idleMs: 7 * DAY_MS } }); + expect(await worker.sweepTerminalWorkspaces()).toMatchObject({ archived: 1 }); + expect(await exists(f.cwd)).toBe(false); + expect(await f.reopen()).toMatchObject({ ok: true, reopened: true }); + expect(await fs.readFile(path.join(f.cwd, "notes", "todo.md"), "utf8")).toBe("untracked work\n"); + }); + + it("counts activity on another issue bound to the workspace", async () => { + const f = await fixture(); + await makeUnsavedWork(f.cwd); + await openIdle(f, 8); + const other = randomUUID(); + await db.insert(issues).values({ id: other, companyId: f.companyId, title: "Unrelated issue sharing the workspace", status: "todo", + executionWorkspaceId: f.workspaceId, updatedAt: new Date(Date.now() - DAY_MS) }); + expect((await gitWorkspaceLastActivityAt(db, await f.row()))!.getTime()).toBeGreaterThan(Date.now() - 2 * DAY_MS); + expect(await f.expire(idle())).toMatchObject({ expired: false, skipReason: "live_or_unresolved_consumer" }); + expect(await exists(f.cwd)).toBe(true); + }); + + it("never expires open issues when idle expiry is off", async () => { + const f = await fixture(); + await makeUnsavedWork(f.cwd); + await openIdle(f, 30); + const worker = executionWorkspaceService(db, { gitCacheExpirationEnabled: true, + gitRestorePoints: { storage: getStorageService(), terminalDelayMs: 3 * DAY_MS, keyDurable: true, idleMs: null } }); + expect(await worker.sweepTerminalWorkspaces()).toMatchObject({ archived: 0, skippedNonTerminalTree: 1 }); + expect(await worker.nextTerminalWorkspaceSweepAt()).toBeNull(); + expect(await exists(f.cwd)).toBe(true); + }); + + it("keeps an open issue's worktree that saw activity inside the idle window", async () => { + const f = await fixture(); + await makeUnsavedWork(f.cwd); + await openIdle(f, 3); + expect(await f.expire(idle())).toMatchObject({ expired: false, skipReason: "live_or_unresolved_consumer" }); + expect(await exists(f.cwd)).toBe(true); + }); + + it("keeps an open issue's worktree whose latest run finished inside the idle window", async () => { + const f = await fixture(); + await makeUnsavedWork(f.cwd); + await openIdle(f, 8); + const agentId = randomUUID(); + await db.insert(agents).values({ id: agentId, companyId: f.companyId, name: "Worker", adapterType: "process" }); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId, issueId: f.issueId, status: "succeeded", + createdAt: new Date(Date.now() - 2 * DAY_MS), startedAt: new Date(Date.now() - 2 * DAY_MS), finishedAt: new Date(Date.now() - 2 * DAY_MS) }); + expect(await gitWorkspaceLastActivityAt(db, await f.row())).toEqual(expect.any(Date)); + expect(await f.expire(idle())).toMatchObject({ expired: false, skipReason: "live_or_unresolved_consumer" }); + expect(await exists(f.cwd)).toBe(true); + }); + }); + it("reports a restore point candidate in dry run without storing anything", async () => { const f = await fixture(); await makeUnsavedWork(f.cwd); diff --git a/server/src/__tests__/heartbeat-git-restore-point-resume.test.ts b/server/src/__tests__/heartbeat-git-restore-point-resume.test.ts new file mode 100644 index 00000000000..dc5ebb6d3f2 --- /dev/null +++ b/server/src/__tests__/heartbeat-git-restore-point-resume.test.ts @@ -0,0 +1,162 @@ +import { execFileSync } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { promises as fs } from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { pathToFileURL } from "node:url"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { + agentWakeupRequests, + agents, + authUsers, + companies, + companyMemberships, + createDb, + EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, + executionWorkspaceRestorePoints, + executionWorkspaces, + heartbeatRuns, + issues, + projects, +} from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { drainHeartbeatRunsToQuiescence } from "./helpers/drain-heartbeat-runs.js"; +import { heartbeatService } from "../services/heartbeat.ts"; +import { instanceSettingsService } from "../services/instance-settings.ts"; +import { materializeIsolatedRemoteWorkspaceSource } from "../services/isolated-remote-workspace-source.js"; +import { realizeExecutionWorkspace } from "../services/workspace-runtime.js"; +import { expireReconstructableGitWorkspace } from "../services/git-workspace-expiration.js"; +import { getStorageService } from "../storage/index.js"; + +const adapterExecute = vi.hoisted(() => vi.fn(async (_input?: unknown) => ({ + exitCode: 0, signal: null, timedOut: false, summary: "Resumed restored work.", provider: "test", model: "test-model", +}))); + +vi.mock("../adapters/index.js", () => ({ + getServerAdapter: () => ({ type: "codex_local", execute: adapterExecute, supportsLocalAgentJwt: false }), + findActiveServerAdapter: () => ({ type: "codex_local", execute: adapterExecute, supportsLocalAgentJwt: false }), + runningProcesses: new Map(), +})); + +const gitEnv = { ...process.env, GIT_AUTHOR_NAME: "Fixture", GIT_AUTHOR_EMAIL: "fixture@example.test", + GIT_COMMITTER_NAME: "Fixture", GIT_COMMITTER_EMAIL: "fixture@example.test" }; +const git = (cwd: string, ...args: string[]) => execFileSync("git", args, { cwd, env: gitEnv, encoding: "utf8" }).trim(); +const DAY_MS = 24 * 60 * 60 * 1000; + +const support = await getEmbeddedPostgresTestSupport(); +describe.skipIf(!support.supported)("heartbeat resume of an idle-expired workspace", () => { + let db: ReturnType; + let database: Awaited>; + let root: string; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("heartbeat-git-restore-point-resume-"); + db = createDb(database.connectionString); + root = await fs.realpath(await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-heartbeat-restore-point-"))); + }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); + afterEach(async () => { + await drainHeartbeatRunsToQuiescence(db, heartbeatService(db)); + adapterExecute.mockClear(); + await db.delete(executionWorkspaceRestorePoints); + await db.execute(sql`TRUNCATE TABLE companies CASCADE`); + await db.delete(authUsers).where(eq(authUsers.id, "responsible-user")); + vi.unstubAllEnvs(); + }); + afterAll(async () => { + await db.$client.end(); + await database.cleanup(); + await fs.rm(root, { recursive: true, force: true }); + }, 60_000); + + async function seed() { + const home = path.join(root, randomUUID()); + vi.stubEnv("PAPERCLIP_HOME", home); + vi.stubEnv("PAPERCLIP_INSTANCE_ID", "default"); + vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "31".repeat(32)); + const companyId = randomUUID(), projectId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(); + const workspaceId = randomUUID(), runId = randomUUID(), wakeupId = randomUUID(); + const issuePrefix = `R${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`; + const location = path.join(home, "repos"), author = path.join(location, "author"), remote = path.join(location, "remote.git"); + await fs.mkdir(author, { recursive: true }); + git(author, "init", "--initial-branch=main"); + await fs.writeFile(path.join(author, "marker.txt"), "commit A\n"); + git(author, "add", "."); git(author, "commit", "-m", "A"); + git(location, "clone", "--bare", author, remote); + const repoUrl = pathToFileURL(remote).href; + await instanceSettingsService(db).updateExperimental({ enableIsolatedWorkspaces: true }); + await db.insert(companies).values({ id: companyId, name: "Restore point resume", issuePrefix, status: "active", defaultResponsibleUserId: "responsible-user" }); + await db.insert(authUsers).values({ id: "responsible-user", name: "Fixture user", email: "responsible-user@example.test", createdAt: new Date(), updatedAt: new Date() }).onConflictDoNothing(); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: "responsible-user", status: "active", membershipRole: "member" }); + await db.insert(projects).values({ id: projectId, companyId, name: "Resumed project", status: "active", + executionWorkspacePolicy: { enabled: true, defaultMode: "isolated_workspace", workspaceStrategy: { type: "git_worktree", baseRef: "origin/main" } } }); + await db.insert(agents).values({ id: agentId, companyId, name: "Resumer", role: "engineer", status: "idle", adapterType: "codex_local", + adapterConfig: {}, runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } }, permissions: {} }); + const idleSince = new Date(Date.now() - 8 * DAY_MS); + await db.insert(issues).values({ id: issueId, companyId, projectId, title: "Stalled work", status: "in_progress", workMode: "standard", + assigneeAgentId: agentId, responsibleUserId: "responsible-user", issueNumber: 1, identifier: `${issuePrefix}-1`, + executionWorkspaceSettings: { mode: "isolated_workspace" }, updatedAt: idleSince }); + const baseCwd = await materializeIsolatedRemoteWorkspaceSource({ companyId, issueId, repoUrl, resolveGitAuth: async () => null }); + const realized = await realizeExecutionWorkspace({ + base: { baseCwd, source: "task_session", projectId, workspaceId: null, repoUrl, repoRef: "origin/main" }, + issue: { id: issueId, identifier: `${issuePrefix}-1`, title: "Stalled work" }, agent: { id: agentId, name: "Resumer", companyId }, + config: { workspaceStrategy: { type: "git_worktree", branchTemplate: "stalled-task", baseRef: "origin/main" } }, skipHostProvision: true, + }); + await db.insert(executionWorkspaces).values({ id: workspaceId, companyId, projectId, sourceIssueId: issueId, mode: "isolated_workspace", + strategyType: "git_worktree", name: "Stalled work", status: "active", providerType: "git_worktree", cwd: realized.cwd, providerRef: realized.cwd, + repoUrl, baseRef: "origin/main", branchName: realized.branchName, + metadata: { createdByRuntime: realized.branchCreatedByRuntime, gitBranchOwnershipVersion: 1, baseRefSha: realized.baseRefSha, + isolatedRemoteSource: { schema: "paperclip.isolated-remote-source.v1", companyId, issueId, repoUrl } } }); + await db.update(issues).set({ executionWorkspaceId: workspaceId, executionWorkspacePreference: "reuse_existing", updatedAt: idleSince }) + .where(eq(issues.id, issueId)); + await fs.writeFile(path.join(realized.cwd, "marker.txt"), "commit B (unpushed)\n"); + git(realized.cwd, "commit", "-qam", "B"); + const unpushed = git(realized.cwd, "rev-parse", "HEAD"); + await fs.writeFile(path.join(realized.cwd, "notes.md"), "uncommitted notes\n"); + const workspace = () => db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspaceId)).then(rows => rows[0]!); + const expiration = await expireReconstructableGitWorkspace(db, await workspace(), { + dryRun: false, expireBase: false, now: new Date(), idleCutoff: new Date(Date.now() - 7 * DAY_MS), + restorePoint: { storage: getStorageService(), notBefore: new Date(Date.now() - DAY_MS), keyDurable: true }, + }); + const queueRun = async () => { + await db.insert(agentWakeupRequests).values({ id: wakeupId, companyId, agentId, source: "assignment", triggerDetail: "system", + reason: "issue_assigned", payload: { issueId }, status: "queued", runId }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "assignment", triggerDetail: "system", status: "queued", + wakeupRequestId: wakeupId, responsibleUserId: "responsible-user", contextSnapshot: { issueId, taskId: issueId, wakeReason: "issue_assigned" } }); + await db.update(issues).set({ checkoutRunId: runId, executionRunId: runId, executionAgentNameKey: "resumer", executionLockedAt: new Date() }) + .where(eq(issues.id, issueId)); + }; + return { issueId, workspaceId, runId, cwd: realized.cwd, unpushed, expiration, workspace, queueRun }; + } + + async function waitForRun(heartbeat: ReturnType, runId: string) { + const deadline = Date.now() + 20_000; + while (Date.now() < deadline) { + const run = await heartbeat.getRun(runId); + if (run && run.status !== "queued" && run.status !== "running") return run; + await new Promise(resolve => setTimeout(resolve, 50)); + } + return heartbeat.getRun(runId); + } + + it("restores the expired workspace and runs the agent in it", async () => { + const seeded = await seed(); + expect(seeded.expiration).toMatchObject({ expired: true, restorePoint: true }); + await expect(fs.access(seeded.cwd)).rejects.toMatchObject({ code: "ENOENT" }); + await seeded.queueRun(); + const heartbeat = heartbeatService(db); + await heartbeat.resumeQueuedRuns(); + const run = await waitForRun(heartbeat, seeded.runId); + await heartbeat.waitForRunExecutionDrain(seeded.runId); + expect(run).toMatchObject({ status: "succeeded" }); + expect(adapterExecute).toHaveBeenCalledTimes(1); + expect(adapterExecute.mock.calls[0]?.[0]).toMatchObject({ context: { executionWorkspaceId: seeded.workspaceId, + paperclipWorkspace: { cwd: seeded.cwd } } }); + expect(git(seeded.cwd, "rev-parse", "HEAD")).toBe(seeded.unpushed); + expect(await fs.readFile(path.join(seeded.cwd, "notes.md"), "utf8")).toBe("uncommitted notes\n"); + const workspace = await seeded.workspace(); + expect(workspace.status).not.toBe("archived"); + expect(workspace.metadata?.reopenPendingConsumption).toBeUndefined(); + const [issue] = await db.select().from(issues).where(eq(issues.id, seeded.issueId)); + expect(issue.executionWorkspaceId).toBe(seeded.workspaceId); + }); +}); diff --git a/server/src/config.ts b/server/src/config.ts index bc974203ec6..05318dd8eaf 100644 --- a/server/src/config.ts +++ b/server/src/config.ts @@ -94,6 +94,7 @@ export interface Config { gitRestorePointRetentionDays: number; gitRestorePointsEnabled: boolean; gitRestorePointTerminalDelayDays: number; + gitRestorePointIdleDays: number; serveUi: boolean; uiDevMiddleware: boolean; secretsProvider: SecretProvider; @@ -342,6 +343,7 @@ export function loadConfig(): Config { const nativeHarnessBackupExpirationDays = Math.max(1, nonNegativeIntegerEnv("PAPERCLIP_NATIVE_HARNESS_BACKUP_EXPIRATION_DAYS", 7)); const gitRestorePointRetentionDays = Math.max(1, nonNegativeIntegerEnv("PAPERCLIP_GIT_RESTORE_POINT_RETENTION_DAYS", 90)); const gitRestorePointTerminalDelayDays = nonNegativeIntegerEnv("PAPERCLIP_GIT_RESTORE_POINT_TERMINAL_DELAY_DAYS", 3); + const gitRestorePointIdleDays = nonNegativeIntegerEnv("PAPERCLIP_GIT_RESTORE_POINT_IDLE_DAYS", 7); const bindValidationErrors = validateConfiguredBindMode({ deploymentMode, deploymentExposure, @@ -402,6 +404,7 @@ export function loadConfig(): Config { gitRestorePointRetentionDays, gitRestorePointsEnabled: process.env.PAPERCLIP_GIT_RESTORE_POINTS_ENABLED === "true", gitRestorePointTerminalDelayDays, + gitRestorePointIdleDays, serveUi: process.env.SERVE_UI !== undefined ? process.env.SERVE_UI === "true" diff --git a/server/src/index.ts b/server/src/index.ts index d8f26a3dbf9..36a66ee0405 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1483,6 +1483,7 @@ async function startServerWithDatabaseTeardown( gitRestorePoints: config.gitRestorePointsEnabled ? { storage: storageService, terminalDelayMs: config.gitRestorePointTerminalDelayDays * 24 * 60 * 60 * 1000, + idleMs: config.gitRestorePointIdleDays > 0 ? config.gitRestorePointIdleDays * 24 * 60 * 60 * 1000 : null, keyDurable: config.storageProvider !== "s3" || Boolean(process.env.PAPERCLIP_SECRETS_MASTER_KEY?.trim()), } : undefined, workspaceReaperCooldownDays: config.workspaceReaperCooldownDays, diff --git a/server/src/services/execution-workspaces.ts b/server/src/services/execution-workspaces.ts index 4c0eb887a41..231fd6c51e7 100644 --- a/server/src/services/execution-workspaces.ts +++ b/server/src/services/execution-workspaces.ts @@ -5,7 +5,7 @@ import { measureRunTransaction } from "./execution-transaction-timing.js"; import { measureRunPhase } from "./execution-preparation-timing.js"; import { assertExecutionWorkspaceRecordAccess, assertTaskWorkspaceExecutionAccess, assertTaskWorkspaceSourceProjectAccess } from "./task-workspace-access.js"; import { verifiedRecordedIsolatedRemoteSource, withIsolatedRepositorySourceLock } from "./isolated-remote-workspace-source.js"; -import { expireReconstructableGitWorkspace, hasPendingGitExpiration } from "./git-workspace-expiration.js"; +import { expireReconstructableGitWorkspace, gitWorkspaceLastActivityAt, hasPendingGitExpiration } from "./git-workspace-expiration.js"; import type { StorageService } from "../storage/types.js"; import { readGitCacheRestoration, restoreGitCacheBase, recoverGitCacheExpirationLocks, GitCacheRestorationFailure } from "./git-workspace-restoration.js"; import { parseObject } from "../adapters/utils.js"; @@ -299,7 +299,7 @@ export type ExecutionWorkspaceServiceOptions = { gitCacheExpirationEnabled?: boolean; gitBaseExpirationEnabled?: boolean; gitCacheExpirationDryRun?: boolean; - gitRestorePoints?: { storage: StorageService; terminalDelayMs: number; keyDurable: boolean }; + gitRestorePoints?: { storage: StorageService; terminalDelayMs: number; keyDurable: boolean; idleMs?: number | null }; terminalWorkspaceInspectionBackoffMs?: number; resolvePullRequestDetails?: PullRequestMergeDetailsResolver; now?: () => Date; @@ -3326,6 +3326,34 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic ), }); } + const idleMs = opts.gitRestorePoints?.idleMs; + const pendingExpiration = opts.gitCacheExpirationEnabled && workspace.providerType !== "plugin" && hasPendingGitExpiration(workspace); + if (pendingExpiration || (sourceIssue && idleMs && !reopenPending && opts.gitCacheExpirationEnabled && workspace.providerType !== "plugin" + && !isClosedExecutionWorkspaceStatus(workspace.status) + && workspace.mode === "isolated_workspace" && workspace.strategyType === "git_worktree")) { + const lastActivityAt = pendingExpiration ? null : await gitWorkspaceLastActivityAt(db, workspace); + const idleDueAt = pendingExpiration ? now().getTime() : (lastActivityAt?.getTime() ?? 0) + idleMs!; + if (idleDueAt <= now().getTime()) { + const expiration = await expireReconstructableGitWorkspace(db, workspace, { + dryRun: opts.gitCacheExpirationDryRun ?? false, expireBase: opts.gitBaseExpirationEnabled ?? false, + now: now(), signal, beforeDelete: opts.beforeTerminalWorkspaceCleanup, + idleCutoff: idleMs ? new Date(now().getTime() - idleMs) : undefined, + restorePoint: opts.gitRestorePoints ? { storage: opts.gitRestorePoints.storage, notBefore: new Date(idleDueAt), + keyDurable: opts.gitRestorePoints.keyDurable } : undefined, + }); + if (expiration.candidate) result.eligible += 1; + if (expiration.expired) result.archived += 1; + const [after] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspace.id)); + if (!expiration.expired || (after && hasPendingGitExpiration(after))) { + terminalWorkspaceDeadlines.set(workspace.id, now().getTime() + 60 * 60 * 1000); + } + logger.info({ event: "execution_workspace.git_cache_expiration", executionWorkspaceId: workspace.id, + companyId: workspace.companyId, dryRun: opts.gitCacheExpirationDryRun ?? false, idle: true, ...expiration }, + "Git workspace cache expiration inspected"); + continue; + } + terminalWorkspaceDeadlines.set(workspace.id, idleDueAt); + } result.skippedNonTerminalTree += 1; continue; } diff --git a/server/src/services/git-workspace-expiration.ts b/server/src/services/git-workspace-expiration.ts index e567b5fbe19..76cfcef5264 100644 --- a/server/src/services/git-workspace-expiration.ts +++ b/server/src/services/git-workspace-expiration.ts @@ -74,25 +74,54 @@ async function restorePointBoundary(cwd: string, repoUrl: string, headOnRemote: } /** Re-read the whole tree and every runtime/recovery consumer immediately before deletion. */ -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` +function expirationScope(workspace: Workspace) { + return sql` WITH RECURSIVE tree AS ( SELECT id, status, completed_at, cancelled_at, updated_at FROM ${issues} WHERE company_id = ${workspace.companyId} AND id = ${workspace.sourceIssueId} UNION SELECT c.id, c.status, c.completed_at, c.cancelled_at, c.updated_at FROM ${issues} c JOIN tree p ON c.parent_id = p.id WHERE c.company_id = ${workspace.companyId} + ), bound AS ( + SELECT id, updated_at FROM ${issues} + WHERE company_id = ${workspace.companyId} AND execution_workspace_id = ${workspace.id} ), relevant_runs AS ( SELECT r.* FROM ${heartbeatRuns} r WHERE r.company_id = ${workspace.companyId} AND ( r.issue_id IN (SELECT id FROM tree) OR r.native_issue_id IN (SELECT id FROM tree) + OR r.issue_id IN (SELECT id FROM bound) OR r.native_issue_id IN (SELECT id FROM bound) OR r.context_snapshot->>'issueId' IN (SELECT id::text FROM tree) OR r.context_snapshot->>'taskId' IN (SELECT id::text FROM tree) OR r.context_snapshot->>'executionWorkspaceId' = ${workspace.id} 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 CASE WHEN + )`; +} + +export async function gitWorkspaceLastActivityAt(db: Db | Tx, workspace: Workspace): Promise { + if (!workspace.sourceIssueId) return null; + const rows = await db.execute<{ at: string | Date | null }>(sql`${expirationScope(workspace)} + SELECT greatest( + (SELECT max(updated_at) FROM tree), + (SELECT max(updated_at) FROM bound), + (SELECT max(greatest(created_at, started_at, finished_at)) FROM relevant_runs) + ) AS at`); + const at = rows[0]?.at; + return at ? new Date(at) : null; +} + +export async function isGitWorkspaceExpirationEligible(tx: Tx, workspace: Workspace, now: Date, idleCutoff?: Date) { + if (workspace.metadata?.reopenPendingConsumption === true) return false; + const closedTree = idleCutoff ? sql` + NOT EXISTS (SELECT 1 FROM tree WHERE updated_at > ${idleCutoff.toISOString()}::timestamptz) + AND NOT EXISTS (SELECT 1 FROM bound WHERE updated_at > ${idleCutoff.toISOString()}::timestamptz) + AND NOT EXISTS (SELECT 1 FROM relevant_runs WHERE greatest(created_at, started_at, finished_at) > ${idleCutoff.toISOString()}::timestamptz)` : sql` + 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) + AND NOT EXISTS (SELECT 1 FROM ${issues} WHERE company_id = ${workspace.companyId} + AND execution_workspace_id = ${workspace.id} AND status NOT IN ('done', 'cancelled'))`; + const rows = await tx.execute<{ allowed: boolean }>(sql`${expirationScope(workspace)} + 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} || '/')) @@ -100,10 +129,7 @@ export async function isGitWorkspaceExpirationEligible(tx: Tx, workspace: Worksp 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) - AND NOT EXISTS (SELECT 1 FROM ${issues} WHERE company_id = ${workspace.companyId} - AND execution_workspace_id = ${workspace.id} AND status NOT IN ('done', 'cancelled')) + AND ${closedTree} AND NOT EXISTS (SELECT 1 FROM relevant_runs WHERE status NOT IN ('succeeded', 'failed', 'timed_out', 'cancelled', 'interrupted') OR finished_at IS NULL OR controller_lease_expires_at > now()) AND NOT EXISTS (SELECT 1 FROM ${issueRecoveryActions} WHERE company_id = ${workspace.companyId} @@ -193,6 +219,7 @@ export async function expireReconstructableGitWorkspace(db: Db, snapshot: Worksp dryRun: boolean; expireBase: boolean; now: Date; signal?: AbortSignal; beforeDelete?: (workspace: Workspace) => Promise; restorePoint?: GitRestorePointExpirationOptions; + idleCutoff?: Date; }): Promise { const result: GitExpirationResult = { candidate: false, expired: false, reclaimedBytes: 0, candidateBytes: 0, baseExpired: false, skipReason: null, restorePoint: false }; @@ -205,7 +232,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 isGitWorkspaceExpirationEligible(tx, workspace, options.now)) throw new Error("live_or_unresolved_consumer"); + if (!await isGitWorkspaceExpirationEligible(tx, workspace, options.now, options.idleCutoff)) 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"); @@ -308,7 +335,7 @@ export async function expireReconstructableGitWorkspace(db: Db, snapshot: Worksp await recoverGitCacheExpirationLocks(workspace!); const lockLocations = gitCacheExpirationLockLocations(workspace!); await options.beforeDelete?.(workspace!); - if (!await isGitWorkspaceExpirationEligible(tx, workspace, options.now)) throw new Error("live_or_unresolved_consumer"); + if (!await isGitWorkspaceExpirationEligible(tx, workspace, options.now, options.idleCutoff)) throw new Error("live_or_unresolved_consumer"); if (!state.workspaceRemoved && present) { const locks = [await acquireGitExpirationLocks(lockLocations.filter(location => location.checkout === null), state.lockToken)]; try { @@ -371,7 +398,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 isGitWorkspaceExpirationEligible(tx, workspace!, options.now)) throw new Error("live_or_unresolved_consumer"); + if (!await isGitWorkspaceExpirationEligible(tx, workspace!, options.now, options.idleCutoff)) 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; diff --git a/server/src/services/heartbeat/environment-selection.test.ts b/server/src/services/heartbeat/environment-selection.test.ts index ce743a16772..cb250773b11 100644 --- a/server/src/services/heartbeat/environment-selection.test.ts +++ b/server/src/services/heartbeat/environment-selection.test.ts @@ -14,6 +14,13 @@ import { buildNativeExecutionInput } from "../native-runtime/native-execution-in import { nativeRuntimeContextFixture } from "../native-runtime/runtime-context.test-fixture.js"; import { selectHeartbeatEnvironment, type HeartbeatEnvironmentSelectionInput } from "./environment-selection.js"; +const restoration = vi.hoisted(() => ({ removed: false })); +vi.mock("../git-workspace-restoration.js", async importOriginal => { + const actual = await importOriginal(); + return { ...actual, readGitCacheExpiration: (workspace: Parameters[0]) => + restoration.removed ? { workspaceRemoved: true } as ReturnType : actual.readGitCacheExpiration(workspace) }; +}); + const support = await getEmbeddedPostgresTestSupport(); describe.skipIf(!support.supported)("heartbeat environment selection boundary", () => { let database: Awaited>; @@ -130,6 +137,56 @@ describe.skipIf(!support.supported)("heartbeat environment selection boundary", expect(result.requestedReusableExecutionWorkspaceConfig).toBeNull(); }); + describe("workspaces removed by git cache expiration", () => { + async function expired() { + const f = await fixture(); + const saved = await f.workspace(); + await executionWorkspaceService(db).update(saved.id, { status: "archived" }); + f.input.task.issueRef!.executionWorkspaceId = saved.id; + f.input.task.issueRef!.executionWorkspacePreference = "reuse_existing"; + await db.update(issues).set({ executionWorkspaceId: saved.id, executionWorkspacePreference: "reuse_existing" }).where(eq(issues.id, f.issue.id)); + restoration.removed = true; + const svc = executionWorkspaceService(db); + f.input.services.executionWorkspacesSvc = svc; + return { ...f, saved, svc }; + } + + it("reopens the issue's expired workspace before the run binds it", async () => { + const { input, issue, saved, svc } = await expired(); + const reopen = vi.spyOn(svc, "reopenClosedIsolatedExecutionWorkspaceForIssue").mockImplementation(async () => { + await executionWorkspaceService(db).update(saved.id, { status: "active" }); + return { ok: true, reopened: true, workspace: (await svc.getStoredById(issue.companyId, saved.id))!, generation: 4 }; + }); + const clear = vi.spyOn(svc, "clearReopenPendingConsumptionForUnconsumedReopen").mockResolvedValue({ cleared: true }); + try { + const result = await selectHeartbeatEnvironment(db, input); + expect(reopen).toHaveBeenCalledWith(expect.objectContaining({ workspaceId: saved.id, + issue: { id: issue.id, companyId: issue.companyId, projectId: issue.projectId }, + authorizationActor: expect.objectContaining({ type: "agent", agentId: input.agent.id, runId: input.run.id }) })); + expect(clear).toHaveBeenCalledWith(expect.objectContaining({ workspaceId: saved.id, expectedGeneration: 4 })); + expect(result.reusableExistingExecutionWorkspace).toMatchObject({ id: saved.id, status: "active" }); + } finally { restoration.removed = false; } + }); + + it("fails the run instead of binding a new workspace when the restore fails", async () => { + const { input, svc } = await expired(); + vi.spyOn(svc, "reopenClosedIsolatedExecutionWorkspaceForIssue") + .mockResolvedValue({ ok: false, code: "rebuild_failed", message: "Workspace restore failed" }); + try { + await expect(selectHeartbeatEnvironment(db, input)).rejects.toThrow("Workspace restore failed"); + } finally { restoration.removed = false; } + }); + + it("leaves a workspace archived for other reasons alone", async () => { + const { input, svc } = await expired(); + restoration.removed = false; + const reopen = vi.spyOn(svc, "reopenClosedIsolatedExecutionWorkspaceForIssue"); + const result = await selectHeartbeatEnvironment(db, input); + expect(reopen).not.toHaveBeenCalled(); + expect(result.reusableExistingExecutionWorkspace).toBeNull(); + }); + }); + it.each(["muse", "openai_dot", "grok_bot"] as const)("keeps %s managed placement without adopting the task's selected workspace", async provider => { const { input, issue, workspace } = await fixture(); const source = await workspace(); diff --git a/server/src/services/heartbeat/environment-selection.ts b/server/src/services/heartbeat/environment-selection.ts index f8e001e5c14..959a352902a 100644 --- a/server/src/services/heartbeat/environment-selection.ts +++ b/server/src/services/heartbeat/environment-selection.ts @@ -1,5 +1,7 @@ import { heartbeatExternalProvider, nativeExecutionUsesTaskWorkspace } from "./external-provider.js"; import { heartbeatRuns, issues, type Db, type agents } from "@paperclipai/db"; +import { and, eq } from "drizzle-orm"; +import { readGitCacheExpiration } from "../git-workspace-restoration.js"; import type { ExecutionWorkspace, IssueExecutionWorkspaceSettings, ProjectExecutionWorkspacePolicy } from "@paperclipai/shared"; import { parseObject } from "../../adapters/utils.js"; import { logger } from "../../middleware/logger.js"; @@ -65,7 +67,8 @@ export interface HeartbeatEnvironmentSelectionInput { issueAssigneeOverrides: { adapterConfig: Record | null } | null; }; services: { - executionWorkspacesSvc: Pick, "getStoredById" | "validateStoredSelectionForRun">; + executionWorkspacesSvc: Pick, "getStoredById" | "validateStoredSelectionForRun" + | "reopenClosedIsolatedExecutionWorkspaceForIssue" | "clearReopenPendingConsumptionForUnconsumedReopen">; environmentsSvc: Pick, "ensureLocalEnvironment" | "findManagedSandboxEnvironment" | "findKubernetesEnvironment" | "ensureKubernetesEnvironment" | "getById">; instanceSettings: Pick, "get" | "getExperimental">; @@ -107,9 +110,32 @@ export async function selectHeartbeatEnvironment(db: Db, input: HeartbeatEnviron const requestedExecutionWorkspaceId = isDotRun ? null : persistedNativeExecutionWorkspaceId ?? readNonEmptyString(issueRef?.executionWorkspaceId); - const existingExecutionWorkspace = requestedExecutionWorkspaceId + let existingExecutionWorkspace = requestedExecutionWorkspaceId ? await executionWorkspacesSvc.getStoredById(agent.companyId, requestedExecutionWorkspaceId) : null; + if (existingExecutionWorkspace && issueId && !isDotRun && !fresh + && existingExecutionWorkspace.id === readNonEmptyString(issueRef?.executionWorkspaceId) + && (existingExecutionWorkspace.status === "archived" || existingExecutionWorkspace.status === "cleanup_failed") + && readGitCacheExpiration(existingExecutionWorkspace as Parameters[0])?.workspaceRemoved) { + const [boundIssue] = await db.select({ id: issues.id, companyId: issues.companyId, projectId: issues.projectId }) + .from(issues).where(and(eq(issues.id, issueId), eq(issues.companyId, agent.companyId))); + if (!boundIssue) throw new Error("Execution workspace issue is unavailable"); + const actor = { agentId: agent.id, actorType: "agent" }; + const reopened = await executionWorkspacesSvc.reopenClosedIsolatedExecutionWorkspaceForIssue({ + workspaceId: existingExecutionWorkspace.id, + issue: boundIssue, + actor, + authorizationActor: { type: "agent", agentId: agent.id, companyId: agent.companyId, source: "agent_jwt", runId: run.id, + onBehalfOfUserId: responsibleUserId === "local-board" ? null : responsibleUserId }, + }); + if (!reopened.ok) throw new Error(reopened.message); + if (reopened.reopened) { + await executionWorkspacesSvc.clearReopenPendingConsumptionForUnconsumedReopen({ + workspaceId: existingExecutionWorkspace.id, issue: boundIssue, actor, expectedGeneration: reopened.generation, + }); + } + existingExecutionWorkspace = await executionWorkspacesSvc.getStoredById(agent.companyId, existingExecutionWorkspace.id); + } const nativeRecoveryExecutionWorkspaceId = resolveNativeRecoveryExecutionWorkspaceBinding({ bindingId: persistedNativeExecutionWorkspaceId,