diff --git a/doc/DATABASE.md b/doc/DATABASE.md index 28950218145..1ff88e3368d 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -62,6 +62,12 @@ the PostgreSQL row version. Native blocked-wait repair compares that version against a fresh full source read before applying changes. Any intervening row update, including an edit outside the projection, defers repair to a fresh sweep. +Execution-blocker reads select process, lifecycle, and task-binding fields rather +than full execution rows. Context retains typed task and pre-dispatch evidence; +Runner profiles retain the adapter type and complete native execution input and +workspace-sync receipts. Full result evidence stays available to the existing +retry and workspace checks. Stored context, profiles, and output are unchanged. + Run-context lookups use expression indexes for both `context_snapshot.issueId` and `context_snapshot.paperclipIssue.id`. Task creation provenance has its own company-scoped expression index that follows the query's native issue ID and diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index fea6694f2e8..e399790fbbb 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -2624,6 +2624,17 @@ Environment overrides: `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_TASK_DIRECTORY_RETENTION_ENABLED=true` lets the terminal workspace reaper + save and remove plain task directories under `/isolated-workspaces/` once + their issue tree has been closed for `PAPERCLIP_GIT_RESTORE_POINT_TERMINAL_DELAY_DAYS`, + and remove orphaned ones (see `doc/recoverable-disk-cleanup.md`). It is off by default + and runs as a dry run until `PAPERCLIP_TASK_DIRECTORY_RETENTION_DRY_RUN=false`. Removal + needs `PAPERCLIP_GIT_RESTORE_POINTS_ENABLED=true`, because the files are saved in an + encrypted restore point first and restored on reopen. + `PAPERCLIP_TASK_DIRECTORY_ORPHAN_GRACE_DAYS=` sets how long an orphaned + directory's issue tree must be closed and its contents unchanged. The default is `7` + and the minimum is `1`. A negative, empty, or non-numeric value falls back to the + default. - `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 ea4aa982001..b33a0cc74fd 100644 --- a/doc/recoverable-disk-cleanup.md +++ b/doc/recoverable-disk-cleanup.md @@ -225,6 +225,131 @@ still keep a worktree from expiring. Restore points are refused when the file storage is `s3` and the secrets master key does not come from `PAPERCLIP_SECRETS_MASTER_KEY`. +## Isolated task directories + +Tasks without a project repository run in a plain (non-Git) directory at +`/isolated-workspaces///`. Its execution workspace row has +strategy `project_primary`, provider `local_fs` and that exact path as `cwd`, in +`isolated_workspace` or `shared_workspace` mode. Agents can leave the only copy of their +work there, such as reports or generated files, because nothing pushes it anywhere. + +Before this cleanup existed nothing removed these directories. The terminal workspace +reaper archives a workspace only after proving delivery through Git (a merged pull request +or a merged branch); for a directory that is not a Git checkout the status inspection fails, +so the row was always skipped as undelivered. A workspace archived by hand keeps its +directory too: artifact cleanup removes a `local_fs` directory only when the row says the +runtime created it, and task directory rows never say so. Deleting them on archive would +discard work silently, so retention saves the files first instead. + +With `PAPERCLIP_TASK_DIRECTORY_RETENTION_ENABLED=true` the terminal workspace reaper +handles an open task directory row once every issue in its tree has been `done` or +`cancelled` for `PAPERCLIP_GIT_RESTORE_POINT_TERMINAL_DELAY_DAYS` (default `3`), the same +delay git restore points use. It registers that moment as its deadline, so a closed tree +costs no checks before then. It removes the directory only when all of these hold, checked +under the issue's source lock and the workspace lifecycle lock: + +- the source issue is still bound to the workspace (`issue_not_bound` otherwise), because + reopening restores through that binding; +- no run of the tree or the workspace is queued, running or unfinished, no native run + finalization is unsettled, no environment lease or runtime service is live, no recovery + action is unresolved and the workspace is not waiting for a reopen to be consumed + (`live_or_unresolved_consumer`); these are the checks git cache expiration uses; +- no other open workspace or project workspace points into the directory (closed rows + for the same issue do not count); +- the directory is a real directory, not a symbolic link (`unsafe_task_directory`). + +It then saves a restore point of the directory: every file with its mode, every +directory and every symbolic link (the link itself; links are never followed). A +`node_modules` directory next to a `package.json` is left out, because installing the +package recreates it. Sockets, devices and other special files keep the directory +(`unsupported_filesystem_entry`), and so does a directory over the restore point limits +(`task_directory_too_many_files`, `task_directory_file_too_large`, +`task_directory_too_large`; 20,000 files, 64 MiB per file, 256 MiB in total). The +captured snapshot is decoded once before anything is removed; a directory whose snapshot +could not be restored stays (`task_directory_not_restorable`). The restore +point uses the git restore point ledger, storage and encryption: it is sealed with a +per-restore-point key wrapped by the instance secrets key, uploaded to file storage under +`git-restore-points///task-directory/`, read back and hash-checked, +and activated in the same transaction that archives the workspace and records the +`isolatedTaskDirectoryRetention` state in its metadata. It follows the git restore point +retention (`PAPERCLIP_GIT_RESTORE_POINT_RETENTION_DAYS`, default `90`), counted from the +later of the tree's close and the removal, so a task that closed long ago still keeps its +files for the full period. Restore points must be enabled (`PAPERCLIP_GIT_RESTORE_POINTS_ENABLED=true`, +`restore_points_disabled` otherwise), and they are refused when the file storage is `s3` +and the secrets master key does not come from `PAPERCLIP_SECRETS_MASTER_KEY`. + +Right before deletion the directory is read again. If anything changed since the capture, +the workspace is made active again, the restore point is dropped and the directory stays +(`task_directory_changed`). Otherwise the directory is renamed to a recorded +`.expired-` name beside it, no longer than the original name, and deleted. The +archive and the removal are recorded as `execution_workspace.task_directory_archived` and +`execution_workspace.task_directory_removed` activity. A removal the +server could not finish is finished by a later sweep. + +Reopening the workspace restores the directory before anything uses it: a comment or an +agent update on an issue bound to a closed `isolated_workspace`, or the next run of the +issue in either mode. The files are written to a `.restoring-` staging directory +beside it, whose name is no longer than the original, verified against the saved digest +and renamed into place. An empty directory in the way is replaced; a non-empty +one that does not match keeps the workspace closed and fails the reopen instead of +overwriting it. After a successful reopen the restore point is superseded and removed; the +next retention saves a new one. When the restore point has already expired, the reopen +starts with an empty directory and posts a system comment on the issue saying the earlier +files are no longer available. + +### Orphaned task directories + +The reaper also walks `isolated-workspaces/` once per server start, at most 8 issue +directories per sweep, continuing from that position in the next sweep. It never follows +symbolic links. For each directory: + +- an open execution workspace pointing at or into it keeps it (the reaper handles that + row as above); +- a closed workspace row that its issue is still bound to is handled like an open row: + the directory is saved in a restore point on that row and removed after the terminal + delay, and reopening restores it (outcome `preserved_and_removed`); +- any other directory is an orphan: its workspace row is missing, or archived or + `cleanup_failed` without the issue bound to it. An orphan is removed when its issue is + gone, or every issue in the tree has been closed for + `PAPERCLIP_TASK_DIRECTORY_ORPHAN_GRACE_DAYS` (default `7`), and nothing in it changed + during that period, and no project workspace, live runtime service, unsettled run, + finalization, live lease or unresolved recovery action of the issue refers to it. Leases +count when they belong to a run of the issue tree, to an issue in the tree, or to any +workspace row of the directory. An + orphan is not saved first: no workspace row can reopen it. If its issue is reopened + later without another workspace, its next run starts in an empty directory. + +After that first walk, the reaper looks only at the directories that changes point to: a +workspace row that becomes archived or `cleanup_failed`, or an issue that becomes `done` or +`cancelled`, sends the reaper to that issue's directory and the directories of its +ancestors, whose trees may have just closed. A directory it kept is checked +again at its own deadline: the end of the grace period or terminal delay, or an hour later +when a lock was busy or a live run, lease, service, finalization or recovery action still +used it. A directory whose issue is open waits for the issue to close. With nothing pending +it registers no work. The walk also deletes leftover `.expired-*` and `.restoring-*` +directories older than an hour; when deleting an orphan's trash fails, another walk is +scheduled an hour later. + +### Controls and logs + +| Environment variable | Default | Effect | +| --- | --- | --- | +| `PAPERCLIP_TASK_DIRECTORY_RETENTION_ENABLED` | `false` | Set `true` to let the reaper handle task directories | +| `PAPERCLIP_TASK_DIRECTORY_RETENTION_DRY_RUN` | `true` | Set `false` to save and remove; a dry run only reports candidates and needs no restore points | +| `PAPERCLIP_TASK_DIRECTORY_ORPHAN_GRACE_DAYS` | `7` | Days an orphan's issue tree must be closed and its contents unchanged before removal (minimum `1`) | + +Each open or pending workspace logs `event: "execution_workspace.task_directory_retention"` +with its outcome (`removed`, `dry_run` or `skipped`), skip reason, candidate bytes and +reclaimed bytes. Each directory the walk inspects logs +`event: "execution_workspace.task_directory_cleanup"` with its outcome (`removed`, +`preserved_and_removed`, `dry_run` or `skipped`), reason and bytes. Each reopen that +restores a directory logs `event: "execution_workspace.task_directory_restore"`. A sweep +saves and removes at most 8 open task directories; the rest continue in the next sweep. A +workspace that was skipped is inspected again only when it or its issue changes, or at its +retry deadline, so skipped workspaces do not use up that limit. +Roll out with the dry run, inspect the candidates and skip reasons, then set the dry run to +`false`. + ## Nested project repository worktrees Extra project repositories that have no configured local folder are no longer full diff --git a/server/src/__tests__/heartbeat-task-directory-restore.test.ts b/server/src/__tests__/heartbeat-task-directory-restore.test.ts new file mode 100644 index 00000000000..be964b6190d --- /dev/null +++ b/server/src/__tests__/heartbeat-task-directory-restore.test.ts @@ -0,0 +1,120 @@ +import { randomUUID } from "node:crypto"; +import { promises as fs } from "node:fs"; +import os from "node:os"; +import path from "node:path"; +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, +} 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 { materializeIsolatedTaskDirectory } from "../services/isolated-task-directory.js"; +import { expireIsolatedTaskDirectory, readTaskDirectoryRetention } from "../services/isolated-task-directory-retention.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 files.", 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 DAY_MS = 24 * 60 * 60 * 1000; + +const support = await getEmbeddedPostgresTestSupport(); +describe.skipIf(!support.supported)("heartbeat resume of a removed task directory", () => { + let db: ReturnType; + let database: Awaited>; + let root: string; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("heartbeat-task-directory-restore-"); + db = createDb(database.connectionString); + root = await fs.realpath(await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-heartbeat-task-directory-"))); + }, 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 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 a removed task directory before the run uses it", async () => { + const mode = "shared_workspace"; + vi.stubEnv("PAPERCLIP_HOME", path.join(root, randomUUID())); + vi.stubEnv("PAPERCLIP_INSTANCE_ID", "default"); + vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "31".repeat(32)); + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), workspaceId = randomUUID(); + const runId = randomUUID(), wakeupId = randomUUID(); + const issuePrefix = `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`; + await db.insert(companies).values({ id: companyId, name: "Task directory 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(agents).values({ id: agentId, companyId, name: "Resumer", role: "engineer", status: "idle", adapterType: "codex_local", + adapterConfig: {}, runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } }, permissions: {} }); + await db.insert(issues).values({ id: issueId, companyId, title: "Finished files", status: "done", workMode: "standard", + assigneeAgentId: agentId, responsibleUserId: "responsible-user", issueNumber: 1, identifier: `${issuePrefix}-1`, + completedAt: new Date(Date.now() - 4 * DAY_MS) }); + const cwd = await materializeIsolatedTaskDirectory({ companyId, issueId }); + await fs.mkdir(path.join(cwd, "out")); + await fs.writeFile(path.join(cwd, "out", "report.md"), "only copy of the report\n"); + await db.insert(executionWorkspaces).values({ id: workspaceId, companyId, sourceIssueId: issueId, mode, strategyType: "project_primary", + name: "Finished files", status: "active", providerType: "local_fs", cwd, metadata: { source: "task_session", createdByRuntime: false } }); + await db.update(issues).set({ executionWorkspaceId: workspaceId, executionWorkspacePreference: "reuse_existing", + executionWorkspaceSettings: { mode } }).where(eq(issues.id, issueId)); + const workspace = () => db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspaceId)).then(rows => rows[0]!); + expect(await expireIsolatedTaskDirectory(db, await workspace(), { dryRun: false, now: new Date(), notBefore: new Date(Date.now() - DAY_MS), + restorePoint: { storage: getStorageService(), keyDurable: true } })).toMatchObject({ expired: true }); + await expect(fs.access(cwd)).rejects.toMatchObject({ code: "ENOENT" }); + + await db.update(issues).set({ status: "in_progress", completedAt: null }).where(eq(issues.id, issueId)); + 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)); + const heartbeat = heartbeatService(db); + await heartbeat.resumeQueuedRuns(); + const run = await waitForRun(heartbeat, runId); + await heartbeat.waitForRunExecutionDrain(runId); + expect(run).toMatchObject({ status: "succeeded" }); + expect(adapterExecute).toHaveBeenCalledTimes(1); + expect(adapterExecute.mock.calls[0]?.[0]).toMatchObject({ context: { executionWorkspaceId: workspaceId, paperclipWorkspace: { cwd } } }); + expect(await fs.readFile(path.join(cwd, "out", "report.md"), "utf8")).toBe("only copy of the report\n"); + const restored = await workspace(); + expect(restored.status).not.toBe("archived"); + expect(readTaskDirectoryRetention(restored)).toBeNull(); + }); +}); diff --git a/server/src/__tests__/isolated-task-directory-retention-config.test.ts b/server/src/__tests__/isolated-task-directory-retention-config.test.ts new file mode 100644 index 00000000000..f9a27b2dc8e --- /dev/null +++ b/server/src/__tests__/isolated-task-directory-retention-config.test.ts @@ -0,0 +1,29 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { loadConfig } from "../config.ts"; + +describe("isolated task directory retention config parsing", () => { + afterEach(() => { + vi.unstubAllEnvs(); + }); + + it("is disabled by default, dry run when enabled, with a seven-day orphan grace", () => { + vi.stubEnv("PAPERCLIP_TASK_DIRECTORY_RETENTION_ENABLED", undefined); + vi.stubEnv("PAPERCLIP_TASK_DIRECTORY_RETENTION_DRY_RUN", undefined); + vi.stubEnv("PAPERCLIP_TASK_DIRECTORY_ORPHAN_GRACE_DAYS", undefined); + expect(loadConfig()).toMatchObject({ taskDirectoryRetentionEnabled: false, taskDirectoryRetentionDryRun: true, + taskDirectoryOrphanGraceDays: 7 }); + }); + + it("accepts explicit settings", () => { + vi.stubEnv("PAPERCLIP_TASK_DIRECTORY_RETENTION_ENABLED", "true"); + vi.stubEnv("PAPERCLIP_TASK_DIRECTORY_RETENTION_DRY_RUN", "false"); + vi.stubEnv("PAPERCLIP_TASK_DIRECTORY_ORPHAN_GRACE_DAYS", "14"); + expect(loadConfig()).toMatchObject({ taskDirectoryRetentionEnabled: true, taskDirectoryRetentionDryRun: false, + taskDirectoryOrphanGraceDays: 14 }); + }); + + it.each([["0", 1], ["", 7], ["-1", 7], ["many", 7]])("clamps the orphan grace for %j", (value, expected) => { + vi.stubEnv("PAPERCLIP_TASK_DIRECTORY_ORPHAN_GRACE_DAYS", value); + expect(loadConfig()).toMatchObject({ taskDirectoryOrphanGraceDays: expected }); + }); +}); diff --git a/server/src/__tests__/isolated-task-directory-retention.test.ts b/server/src/__tests__/isolated-task-directory-retention.test.ts new file mode 100644 index 00000000000..bdaa1d97582 --- /dev/null +++ b/server/src/__tests__/isolated-task-directory-retention.test.ts @@ -0,0 +1,524 @@ +import { promises as fs } from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { randomUUID } from "node:crypto"; +import { execFileSync } from "node:child_process"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { eq, sql } from "drizzle-orm"; +import { + activityLog, + agents, + companies, + createDb, + environmentLeases, + executionWorkspaceRestorePoints, + executionWorkspaces, + heartbeatRuns, + issueComments, + issueRecoveryActions, + issues, + nativeRunFinalizations, + workspaceRuntimeServices, +} from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { executionWorkspaceService } from "../services/execution-workspaces.js"; +import { materializeIsolatedTaskDirectory } from "../services/isolated-task-directory.js"; +import { + expireIsolatedTaskDirectory, + readTaskDirectoryRetention, + sweepIsolatedTaskDirectories, + TASK_DIRECTORY_RESTORE_POINT_EXPIRED_WARNING, +} from "../services/isolated-task-directory-retention.js"; +import { markGitRestorePointExpired } from "../services/git-restore-point-storage.js"; +import { gitRestorePointRetentionService } from "../services/git-restore-point-ledger.js"; +import { getStorageService } from "../storage/index.js"; + +const DAY_MS = 24 * 60 * 60 * 1000; +const exists = (target: string) => fs.lstat(target).then(() => true, () => false); + +async function age(target: string, at: Date) { + const stat = await fs.lstat(target); + if (stat.isDirectory()) for (const name of await fs.readdir(target)) await age(path.join(target, name), at); + await (stat.isSymbolicLink() ? fs.lutimes(target, at, at) : fs.utimes(target, at, at)); +} + +const support = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = support.supported ? describe : describe.skip; +describeEmbeddedPostgres("isolated task directory retention", () => { + let db: ReturnType; + let database: Awaited>; + let root: string; + let outside: string; + const previousHome = process.env.PAPERCLIP_HOME; + const previousInstance = process.env.PAPERCLIP_INSTANCE_ID; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("isolated-task-directory-retention-"); + db = createDb(database.connectionString); + root = await fs.realpath(await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-task-directory-retention-"))); + process.env.PAPERCLIP_HOME = path.join(root, "home"); + process.env.PAPERCLIP_INSTANCE_ID = "default"; + vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "31".repeat(32)); + outside = path.join(root, "outside"); + await fs.mkdir(outside); + await fs.writeFile(path.join(outside, "keep.txt"), "outside\n"); + }); + afterEach(async () => { + await db.delete(executionWorkspaceRestorePoints); + await db.execute(sql`TRUNCATE TABLE companies CASCADE`); + await fs.rm(path.join(root, "home", "instances", "default", "isolated-workspaces"), { recursive: true, force: true }); + }); + afterAll(async () => { + vi.unstubAllEnvs(); + if (previousHome === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = previousHome; + if (previousInstance === undefined) delete process.env.PAPERCLIP_INSTANCE_ID; else process.env.PAPERCLIP_INSTANCE_ID = previousInstance; + await db.$client.end({ timeout: 1 }); await database.cleanup(); await fs.rm(root, { recursive: true, force: true }); + }); + + const settings = { dryRun: false, terminalDelayMs: 3 * DAY_MS, orphanGraceMs: 7 * DAY_MS }; + const restorePoints = (overrides: Partial<{ keyDurable: boolean }> = {}) => ({ storage: getStorageService(), terminalDelayMs: 3 * DAY_MS, keyDurable: true, ...overrides }); + + async function fixture(options: { closedDaysAgo?: number | null; mode?: string; status?: string } = {}) { + const companyId = randomUUID(), issueId = randomUUID(), workspaceId = randomUUID(), agentId = randomUUID(); + const closedDaysAgo = options.closedDaysAgo === undefined ? 4 : options.closedDaysAgo; + await db.insert(companies).values({ id: companyId, name: "Task directory fixture", issuePrefix: `TD${companyId.slice(0, 6)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Fixture agent" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Finished task", + status: closedDaysAgo === null ? "in_progress" : "done", completedAt: closedDaysAgo === null ? null : new Date(Date.now() - closedDaysAgo * DAY_MS) }); + const cwd = await materializeIsolatedTaskDirectory({ companyId, issueId }); + await fs.writeFile(path.join(cwd, "report.md"), "final report\n"); + await fs.mkdir(path.join(cwd, "nested", "deeper"), { recursive: true }); + await fs.writeFile(path.join(cwd, "nested", "deeper", "data.json"), "{\"value\":1}\n"); + await fs.writeFile(path.join(cwd, "run.sh"), "#!/bin/sh\necho ok\n", { mode: 0o755 }); + await fs.symlink(outside, path.join(cwd, "outside-link")); + await fs.writeFile(path.join(cwd, "package.json"), "{\"name\":\"fixture\"}\n"); + await fs.mkdir(path.join(cwd, "node_modules", "dep"), { recursive: true }); + await fs.writeFile(path.join(cwd, "node_modules", "dep", "index.js"), "module.exports = 1;\n"); + await db.insert(executionWorkspaces).values({ id: workspaceId, companyId, sourceIssueId: issueId, mode: options.mode ?? "isolated_workspace", + strategyType: "project_primary", name: "Fixture", status: options.status ?? "active", providerType: "local_fs", cwd, + closedAt: options.status === "archived" ? new Date(Date.now() - closedDaysAgo! * DAY_MS) : null, + metadata: { source: "task_session", createdByRuntime: false } }); + await db.update(issues).set({ executionWorkspaceId: workspaceId }).where(eq(issues.id, issueId)); + const row = () => db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspaceId)).then(rows => rows[0]!); + const ledger = () => db.select().from(executionWorkspaceRestorePoints).where(eq(executionWorkspaceRestorePoints.executionWorkspaceId, workspaceId)); + const service = (overrides: Parameters[1] = {}) => executionWorkspaceService(db, { + taskDirectoryRetention: settings, gitRestorePoints: restorePoints(), ...overrides }); + const expire = async (overrides: Partial[2]> = {}) => expireIsolatedTaskDirectory(db, await row(), { + dryRun: false, now: new Date(), notBefore: new Date(Date.now() - DAY_MS), restorePoint: { storage: getStorageService(), keyDurable: true }, ...overrides }); + const reopen = () => executionWorkspaceService(db).reopenClosedIsolatedExecutionWorkspaceForIssue({ workspaceId, + issue: { id: issueId, companyId, projectId: null }, actor: { actorType: "user", agentId: null }, + authorizationActor: { type: "board", source: "local_implicit", isInstanceAdmin: true } }); + return { companyId, issueId, workspaceId, agentId, cwd, row, ledger, service, expire, reopen }; + } + + async function expectRestored(cwd: string) { + expect(await fs.readFile(path.join(cwd, "report.md"), "utf8")).toBe("final report\n"); + expect(await fs.readFile(path.join(cwd, "nested", "deeper", "data.json"), "utf8")).toBe("{\"value\":1}\n"); + expect((await fs.stat(path.join(cwd, "run.sh"))).mode & 0o777).toBe(0o755); + expect((await fs.lstat(path.join(cwd, "outside-link"))).isSymbolicLink()).toBe(true); + expect(await fs.readlink(path.join(cwd, "outside-link"))).toBe(outside); + expect(await exists(path.join(cwd, "package.json"))).toBe(true); + expect(await exists(path.join(cwd, "node_modules"))).toBe(false); + } + + it("archives and removes a closed issue's task directory after the delay and restores it on reopen", async () => { + const f = await fixture(); + const result = await f.service().sweepTerminalWorkspaces(); + expect(result).toMatchObject({ archived: 1 }); + expect(await exists(f.cwd)).toBe(false); + expect(await fs.readFile(path.join(outside, "keep.txt"), "utf8")).toBe("outside\n"); + const archived = await f.row(); + expect(archived).toMatchObject({ status: "archived", cleanupReason: "isolated task directory retention" }); + const state = readTaskDirectoryRetention(archived)!; + expect(state).toMatchObject({ removed: true, restorePoint: { files: 4, excluded: 1 } }); + expect(await f.ledger()).toMatchObject([{ status: "active", objectKey: state.restorePoint.objectKey }]); + + expect(await f.reopen()).toMatchObject({ ok: true, reopened: true }); + await expectRestored(f.cwd); + const reopened = await f.row(); + expect(reopened.status).toBe("active"); + expect(readTaskDirectoryRetention(reopened)).toBeNull(); + expect(await f.ledger()).toMatchObject([{ status: "superseded" }]); + }); + + it("restores a shared-workspace task directory through reopen", async () => { + const f = await fixture({ mode: "shared_workspace" }); + expect(await f.expire()).toMatchObject({ expired: true, restorePoint: true }); + expect(await exists(f.cwd)).toBe(false); + const [issue] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + expect(issue.executionWorkspaceId).toBe(f.workspaceId); + expect(await f.reopen()).toMatchObject({ ok: true, reopened: true }); + await expectRestored(f.cwd); + }); + + it("keeps the directory until the terminal delay and registers that deadline", async () => { + const f = await fixture({ closedDaysAgo: 1 }); + const service = f.service(); + expect(await service.sweepTerminalWorkspaces()).toMatchObject({ archived: 0 }); + expect(await exists(f.cwd)).toBe(true); + expect((await f.row()).status).toBe("active"); + const [issue] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + const deadline = await service.nextTerminalWorkspaceSweepAt(); + expect(deadline).toBe(issue.completedAt!.getTime() + 3 * DAY_MS); + expect(await f.ledger()).toEqual([]); + }); + + it("keeps the directory while the issue is open", async () => { + const f = await fixture({ closedDaysAgo: null }); + expect(await f.service().sweepTerminalWorkspaces()).toMatchObject({ archived: 0, skippedNonTerminalTree: 1 }); + expect(await exists(f.cwd)).toBe(true); + expect(await f.ledger()).toEqual([]); + }); + + it("does nothing when retention is not configured", async () => { + const f = await fixture(); + await executionWorkspaceService(db, { gitRestorePoints: restorePoints() }).sweepTerminalWorkspaces(); + expect(await exists(f.cwd)).toBe(true); + expect((await f.row()).status).toBe("active"); + }); + + type Consumer = (f: Awaited>) => Promise; + const consumers: Array<[string, Consumer, string]> = [ + ["a live run", f => db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running", contextSnapshot: { issueId: f.issueId } }), + "live_or_unresolved_consumer"], + ["a live runtime service", f => db.insert(workspaceRuntimeServices).values({ id: randomUUID(), companyId: f.companyId, scopeType: "execution_workspace", + serviceName: "web", status: "running", lifecycle: "shared", provider: "local_process", executionWorkspaceId: f.workspaceId, cwd: f.cwd }), + "live_or_unresolved_consumer"], + ["a reopen fence", f => db.update(executionWorkspaces).set({ metadata: { source: "task_session", reopenPendingConsumption: true } }) + .where(eq(executionWorkspaces.id, f.workspaceId)), "live_or_unresolved_consumer"], + ["an unsettled finalization", async f => { + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: runId, companyId: f.companyId, agentId: f.agentId, status: "succeeded", + finishedAt: new Date(Date.now() - 4 * DAY_MS), nativeIssueId: f.issueId }); + await db.insert(nativeRunFinalizations).values({ runId, companyId: f.companyId, issueId: f.issueId, phase: "observed", leaseOwner: "native-finalizer:live" }); + }, "live_or_unresolved_consumer"], + ["a live environment lease", f => db.insert(environmentLeases).values({ companyId: f.companyId, executionWorkspaceId: f.workspaceId, status: "active" }), + "live_or_unresolved_consumer"], + ["an unresolved recovery action", f => db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: f.issueId, + kind: "active_run_watchdog", status: "active", ownerType: "board", ownerAgentId: null, cause: "native_workspace_finalization_owner_unverified", + fingerprint: randomUUID(), evidence: {}, nextAction: "Verify the owner." }), "live_or_unresolved_consumer"], + ["an issue bound to another workspace", f => db.update(issues).set({ executionWorkspaceId: null }).where(eq(issues.id, f.issueId)), "issue_not_bound"], + ["a sibling live workspace in the directory", f => db.insert(executionWorkspaces).values({ companyId: f.companyId, mode: "shared_workspace", + strategyType: "project_primary", name: "Sibling", status: "active", providerType: "local_fs", cwd: path.join(f.cwd, "nested") }), + "live_or_unresolved_consumer"], + ]; + + it.each(consumers)("keeps the directory while %s exists", async (_label, add, reason) => { + const f = await fixture(); + await add(f); + expect(await f.expire()).toMatchObject({ expired: false, skipReason: reason }); + expect(await exists(f.cwd)).toBe(true); + expect((await f.row()).status).toBe("active"); + expect(await f.ledger()).toEqual([]); + }); + + it("removes the directory despite closed sibling rows at the same path", async () => { + const f = await fixture(); + await db.insert(executionWorkspaces).values({ companyId: f.companyId, sourceIssueId: f.issueId, mode: "isolated_workspace", + strategyType: "project_primary", name: "Old", status: "archived", closedAt: new Date(Date.now() - 10 * DAY_MS), providerType: "local_fs", cwd: f.cwd }); + expect(await f.expire()).toMatchObject({ expired: true }); + }); + + it("refuses to remove without restore points or with a non-durable key", async () => { + const f = await fixture(); + expect(await f.expire({ restorePoint: undefined })).toMatchObject({ expired: false, skipReason: "restore_points_disabled" }); + expect(await f.expire({ restorePoint: { storage: getStorageService(), keyDurable: false } })) + .toMatchObject({ expired: false, skipReason: "restore_point_key_not_durable" }); + expect(await f.expire({ notBefore: new Date(Date.now() + DAY_MS) })).toMatchObject({ expired: false, skipReason: "not_due" }); + expect(await exists(f.cwd)).toBe(true); + expect(await f.ledger()).toEqual([]); + }); + + it("keeps files it cannot save", async () => { + const f = await fixture(); + expect(await f.expire({ restorePoint: { storage: getStorageService(), keyDurable: true, limits: { maxFiles: 2, maxFileBytes: 1024, maxTotalBytes: 4096 } } })) + .toMatchObject({ expired: false, skipReason: "task_directory_too_many_files" }); + expect(await exists(f.cwd)).toBe(true); + expect((await f.row()).status).toBe("active"); + expect(await f.ledger()).toEqual([]); + }); + + it("reports candidates in dry run without changing anything", async () => { + const f = await fixture(); + const result = await f.service({ taskDirectoryRetention: { ...settings, dryRun: true }, gitRestorePoints: undefined }).sweepTerminalWorkspaces(); + expect(result).toMatchObject({ eligible: 1, archived: 0 }); + expect(await exists(f.cwd)).toBe(true); + expect((await f.row()).status).toBe("active"); + expect(await f.ledger()).toEqual([]); + }); + + it("rolls back when the directory changes after it was saved", async () => { + const f = await fixture(); + const result = await f.expire({ beforeDelete: async () => { await fs.writeFile(path.join(f.cwd, "late.txt"), "late work\n"); } }); + expect(result, JSON.stringify(result.failure)).toMatchObject({ expired: false, skipReason: "task_directory_changed" }); + expect(await fs.readFile(path.join(f.cwd, "late.txt"), "utf8")).toBe("late work\n"); + const row = await f.row(); + expect(row.status).toBe("active"); + expect(readTaskDirectoryRetention(row)).toBeNull(); + expect(await f.ledger()).toMatchObject([{ status: "superseded" }]); + }); + + it("finishes an interrupted removal on the next sweep", async () => { + const f = await fixture(); + await expect(f.expire({ beforeDelete: async () => { throw new Error("simulated_crash"); } })) + .resolves.toMatchObject({ expired: false, skipReason: "simulated_crash" }); + expect(await exists(f.cwd)).toBe(true); + expect(readTaskDirectoryRetention(await f.row())).toMatchObject({ removed: false }); + expect(await f.service().sweepTerminalWorkspaces()).toMatchObject({ archived: 1 }); + expect(await exists(f.cwd)).toBe(false); + expect(readTaskDirectoryRetention(await f.row())).toMatchObject({ removed: true }); + }); + + it("never follows a task directory that is a symbolic link", async () => { + const f = await fixture(); + const moved = path.join(root, `moved-${f.issueId}`); + await fs.rename(f.cwd, moved); + await fs.symlink(moved, f.cwd); + expect(await f.expire()).toMatchObject({ expired: false, skipReason: "unsafe_task_directory" }); + expect(await fs.readFile(path.join(moved, "report.md"), "utf8")).toBe("final report\n"); + expect(await f.ledger()).toEqual([]); + await fs.rm(f.cwd); + await fs.rm(moved, { recursive: true }); + }); + + it("reopens an empty directory with a warning when the saved files expired", async () => { + const f = await fixture(); + expect(await f.expire()).toMatchObject({ expired: true }); + const [ledger] = await f.ledger(); + await markGitRestorePointExpired(db, ledger!); + expect(readTaskDirectoryRetention(await f.row())?.restorePoint.expiredAt).toEqual(expect.any(String)); + expect(await f.reopen()).toMatchObject({ ok: true, reopened: true, warning: TASK_DIRECTORY_RESTORE_POINT_EXPIRED_WARNING }); + expect(await fs.readdir(f.cwd)).toEqual([]); + const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, f.issueId)); + expect(comments.map(comment => comment.body)).toContain(TASK_DIRECTORY_RESTORE_POINT_EXPIRED_WARNING); + }); + + it("refuses to restore over a directory that diverged after removal", async () => { + const f = await fixture(); + expect(await f.expire()).toMatchObject({ expired: true }); + await fs.mkdir(f.cwd); + await fs.writeFile(path.join(f.cwd, "other.txt"), "unrelated\n"); + expect(await f.reopen()).toMatchObject({ ok: false, code: "rebuild_failed" }); + expect(await fs.readFile(path.join(f.cwd, "other.txt"), "utf8")).toBe("unrelated\n"); + expect((await f.row()).status).toBe("archived"); + }); + + it("saves and restores file names that contain a backslash", async () => { + const f = await fixture(); + await fs.writeFile(path.join(f.cwd, "report\\final.txt"), "backslash name\n"); + expect(await f.expire()).toMatchObject({ expired: true }); + expect(await f.reopen()).toMatchObject({ ok: true, reopened: true }); + expect(await fs.readFile(path.join(f.cwd, "report\\final.txt"), "utf8")).toBe("backslash name\n"); + }); + + it("restores a deep path that fills the platform path length", async () => { + const f = await fixture(); + let deep = f.cwd; + while (deep.length < 1000) deep = path.join(deep, "d".repeat(Math.min(60, 1000 - deep.length))); + await fs.mkdir(deep, { recursive: true }); + const file = path.join(deep, "z".repeat(Math.max(1, 1015 - deep.length))); + await fs.writeFile(file, "deep\n"); + expect(await f.expire()).toMatchObject({ expired: true }); + expect(await f.reopen()).toMatchObject({ ok: true, reopened: true }); + expect(await fs.readFile(file, "utf8")).toBe("deep\n"); + }); + + it("keeps an expiry recorded while the directory was being removed, and records activity", async () => { + const f = await fixture(); + const result = await f.expire({ beforeDelete: async () => { + const [ledger] = await f.ledger(); + await markGitRestorePointExpired(db, ledger!); + } }); + expect(result).toMatchObject({ expired: true }); + expect(readTaskDirectoryRetention(await f.row())).toMatchObject({ removed: true, restorePoint: { expiredAt: expect.any(String) } }); + const actions = await db.select({ action: activityLog.action }).from(activityLog).where(eq(activityLog.entityId, f.workspaceId)); + expect(actions.map(row => row.action).sort()).toEqual(["execution_workspace.task_directory_archived", "execution_workspace.task_directory_removed"]); + }); + + it("keeps a new restore point for the full retention even when the task closed long ago", async () => { + const f = await fixture({ closedDaysAgo: 120 }); + expect(await f.expire()).toMatchObject({ expired: true }); + const retention = gitRestorePointRetentionService(db, { storage: getStorageService(), retentionDays: 90 }); + await retention.tick(); + expect(await f.ledger()).toMatchObject([{ status: "active" }]); + expect(readTaskDirectoryRetention(await f.row())?.restorePoint.expiredAt).toBeUndefined(); + expect(await f.reopen()).toMatchObject({ ok: true, reopened: true }); + await expectRestored(f.cwd); + }); + + it("does not let skipped rows starve later rows of the per-sweep budget", async () => { + const blocked = await fixture(); + execFileSync("mkfifo", [path.join(blocked.cwd, "pipe")]); + const ready = await fixture(); + await db.update(executionWorkspaces).set({ updatedAt: new Date(Date.now() - 60_000) }).where(eq(executionWorkspaces.id, blocked.workspaceId)); + const service = blocked.service({ taskDirectoryRetention: { ...settings, perSweep: 1 } }); + expect(await service.sweepTerminalWorkspaces()).toMatchObject({ archived: 0 }); + expect(await exists(ready.cwd)).toBe(true); + expect(await service.sweepTerminalWorkspaces()).toMatchObject({ archived: 1 }); + expect(await exists(ready.cwd)).toBe(false); + expect(await exists(blocked.cwd)).toBe(true); + }); + + describe("folder sweep", () => { + const sweep = (overrides: Partial[1]> = {}) => sweepIsolatedTaskDirectories(db, { + dryRun: false, now: new Date(), orphanGraceMs: 7 * DAY_MS, terminalDelayMs: 3 * DAY_MS, + restorePoint: { storage: getStorageService(), keyDurable: true }, ...overrides }); + + async function orphan(options: { modifiedDaysAgo?: number } = {}) { + const companyId = randomUUID(), issueId = randomUUID(); + const cwd = await materializeIsolatedTaskDirectory({ companyId, issueId }); + await fs.writeFile(path.join(cwd, "left.txt"), "left behind\n"); + await fs.symlink(outside, path.join(cwd, "outside-link")); + await age(cwd, new Date(Date.now() - (options.modifiedDaysAgo ?? 10) * DAY_MS)); + return { companyId, issueId, cwd }; + } + + it("removes an orphan whose issue is gone after the grace period", async () => { + const o = await orphan(); + const batch = await sweep(); + expect(batch.results).toMatchObject([{ issueId: o.issueId, outcome: "removed", reason: null }]); + expect(batch.results[0]!.reclaimedBytes).toBeGreaterThan(0); + expect(await exists(o.cwd)).toBe(false); + expect(await fs.readFile(path.join(outside, "keep.txt"), "utf8")).toBe("outside\n"); + }); + + it("keeps a recently modified orphan and schedules its deadline", async () => { + const o = await orphan({ modifiedDaysAgo: 2 }); + const batch = await sweep(); + expect(batch.results).toMatchObject([{ issueId: o.issueId, outcome: "skipped", reason: "recently_modified" }]); + expect(batch.results[0]!.retryAt).toBeGreaterThan(Date.now() + 4 * DAY_MS); + expect(await exists(o.cwd)).toBe(true); + }); + + it("keeps an orphan in dry run", async () => { + const o = await orphan(); + const batch = await sweep({ dryRun: true }); + expect(batch.results).toMatchObject([{ issueId: o.issueId, outcome: "dry_run" }]); + expect(await exists(o.cwd)).toBe(true); + }); + + it("keeps a folder that a live workspace points into", async () => { + const o = await orphan(); + await db.insert(companies).values({ id: o.companyId, name: "Live", issuePrefix: `LV${o.companyId.slice(0, 6)}` }); + await db.insert(executionWorkspaces).values({ companyId: o.companyId, mode: "shared_workspace", strategyType: "project_primary", + name: "Live", status: "active", providerType: "local_fs", cwd: path.join(o.cwd, "sub") }); + const batch = await sweep(); + expect(batch.results).toEqual([]); + expect(await exists(o.cwd)).toBe(true); + }); + + it("removes a folder whose archived row is unbound and whose issue closed long ago", async () => { + const o = await orphan(); + await db.insert(companies).values({ id: o.companyId, name: "Archived", issuePrefix: `AR${o.companyId.slice(0, 6)}` }); + await db.insert(issues).values({ id: o.issueId, companyId: o.companyId, title: "Old", status: "done", completedAt: new Date(Date.now() - 10 * DAY_MS) }); + await db.insert(executionWorkspaces).values({ companyId: o.companyId, sourceIssueId: o.issueId, mode: "shared_workspace", strategyType: "project_primary", + name: "Archived", status: "archived", closedAt: new Date(Date.now() - 10 * DAY_MS), providerType: "local_fs", cwd: o.cwd }); + expect((await sweep()).results).toMatchObject([{ outcome: "removed" }]); + expect(await exists(o.cwd)).toBe(false); + }); + + it("keeps an orphan whose issue is still open", async () => { + const o = await orphan(); + await db.insert(companies).values({ id: o.companyId, name: "Open", issuePrefix: `OP${o.companyId.slice(0, 6)}` }); + await db.insert(issues).values({ id: o.issueId, companyId: o.companyId, title: "Open", status: "todo" }); + expect((await sweep()).results).toMatchObject([{ outcome: "skipped", reason: "issue_open" }]); + expect(await exists(o.cwd)).toBe(true); + }); + + it("never follows an orphan that is a symbolic link", async () => { + const companyId = randomUUID(), issueId = randomUUID(); + const parent = path.dirname(await materializeIsolatedTaskDirectory({ companyId, issueId })); + await fs.rmdir(path.join(parent, issueId)); + await fs.symlink(outside, path.join(parent, issueId)); + expect((await sweep()).results).toMatchObject([{ outcome: "skipped", reason: "unsafe_task_directory" }]); + expect(await fs.readFile(path.join(outside, "keep.txt"), "utf8")).toBe("outside\n"); + }); + + it("preserves an archived bound workspace's surviving folder before removing it, and reopen restores it", async () => { + const f = await fixture({ status: "archived" }); + const batch = await sweep(); + expect(batch.results).toMatchObject([{ issueId: f.issueId, outcome: "preserved_and_removed", executionWorkspaceId: f.workspaceId }]); + expect(await exists(f.cwd)).toBe(false); + expect(await f.ledger()).toMatchObject([{ status: "active" }]); + expect(await f.reopen()).toMatchObject({ ok: true, reopened: true }); + await expectRestored(f.cwd); + }); + + it("retries an orphan that a live run still uses", async () => { + const o = await orphan(); + const agentId = randomUUID(); + await db.insert(companies).values({ id: o.companyId, name: "Busy", issuePrefix: `BS${o.companyId.slice(0, 6)}` }); + await db.insert(agents).values({ id: agentId, companyId: o.companyId, name: "Busy agent" }); + await db.insert(issues).values({ id: o.issueId, companyId: o.companyId, title: "Old", status: "done", completedAt: new Date(Date.now() - 10 * DAY_MS) }); + await db.insert(heartbeatRuns).values({ companyId: o.companyId, agentId, status: "running", contextSnapshot: { issueId: o.issueId } }); + const batch = await sweep(); + expect(batch.results).toMatchObject([{ outcome: "skipped", reason: "live_run" }]); + expect(batch.results[0]!.retryAt).toBeGreaterThan(Date.now()); + expect(await exists(o.cwd)).toBe(true); + }); + + it("finds a folder left by a later archive through workspace change hints", async () => { + const service = executionWorkspaceService(db, { taskDirectoryRetention: settings, gitRestorePoints: restorePoints() }); + await service.sweepTerminalWorkspaces(); + expect(await service.nextTerminalWorkspaceSweepAt()).toBeNull(); + const f = await fixture({ status: "archived" }); + await service.sweepTerminalWorkspaces(); + expect(await exists(f.cwd)).toBe(true); + service.terminalWorkspaceWorkSettled(new Set([`workspace:${f.workspaceId}`])); + expect(await service.nextTerminalWorkspaceSweepAt()).toBeLessThanOrEqual(Date.now()); + await service.sweepTerminalWorkspaces(); + expect(await exists(f.cwd)).toBe(false); + expect(readTaskDirectoryRetention(await f.row())).toMatchObject({ removed: true }); + expect(await service.nextTerminalWorkspaceSweepAt()).toBeNull(); + }); + + it("keeps overflowing folder deadlines discoverable through another walk", async () => { + const folders = [await orphan({ modifiedDaysAgo: 1 }), await orphan({ modifiedDaysAgo: 1 })]; + let clock = Date.now(); + const service = executionWorkspaceService(db, { taskDirectoryRetention: { ...settings, trackedLimit: 1 }, + gitRestorePoints: restorePoints(), now: () => new Date(clock) }); + await service.sweepTerminalWorkspaces(); + for (const folder of folders) expect(await exists(folder.cwd)).toBe(true); + const deadline = await service.nextTerminalWorkspaceSweepAt(); + expect(deadline).toBeGreaterThan(clock); + clock = deadline! + 1000; + await service.sweepTerminalWorkspaces(); + for (const folder of folders) expect(await exists(folder.cwd)).toBe(false); + }); + + it("returns to a parent's folder when its last open child closes", async () => { + const service = executionWorkspaceService(db, { taskDirectoryRetention: settings, gitRestorePoints: restorePoints() }); + const f = await fixture({ status: "archived" }); + const childId = randomUUID(); + await db.insert(issues).values({ id: childId, companyId: f.companyId, parentId: f.issueId, title: "Open child", status: "todo" }); + await service.sweepTerminalWorkspaces(); + expect(await exists(f.cwd)).toBe(true); + await db.update(issues).set({ status: "done", completedAt: new Date(Date.now() - 4 * DAY_MS) }).where(eq(issues.id, childId)); + service.terminalWorkspaceWorkSettled(new Set([`issue:${childId}`])); + await service.sweepTerminalWorkspaces(); + expect(await exists(f.cwd)).toBe(false); + expect(readTaskDirectoryRetention(await f.row())).toMatchObject({ removed: true }); + }); + + it("keeps an orphan whose closed workspace still has a lease cleanup pending", async () => { + const o = await orphan(); + await db.insert(companies).values({ id: o.companyId, name: "Lease", issuePrefix: `LS${o.companyId.slice(0, 6)}` }); + const [row] = await db.insert(executionWorkspaces).values({ companyId: o.companyId, mode: "shared_workspace", strategyType: "project_primary", + name: "Closed", status: "archived", closedAt: new Date(Date.now() - 10 * DAY_MS), providerType: "local_fs", cwd: o.cwd }).returning(); + await db.insert(environmentLeases).values({ companyId: o.companyId, executionWorkspaceId: row!.id, status: "pending_cleanup", + releasedAt: new Date(Date.now() - 10 * DAY_MS) }); + expect((await sweep()).results).toMatchObject([{ outcome: "skipped", reason: "live_lease" }]); + expect(await exists(o.cwd)).toBe(true); + }); + + it("bounds the work per sweep and continues from its cursor", async () => { + const folders = [await orphan(), await orphan(), await orphan()]; + const first = await sweep({ limit: 2 }); + expect(first.results).toHaveLength(2); + expect(first.next).not.toBeNull(); + const second = await sweep({ limit: 2, after: first.next }); + expect(second.results).toHaveLength(1); + expect(second.next).toBeNull(); + for (const folder of folders) expect(await exists(folder.cwd)).toBe(false); + }); + }); +}); diff --git a/server/src/config.ts b/server/src/config.ts index e63b66ee0b0..74e80c77ee1 100644 --- a/server/src/config.ts +++ b/server/src/config.ts @@ -102,6 +102,9 @@ export interface Config { gitRestorePointsEnabled: boolean; gitRestorePointTerminalDelayDays: number; gitRestorePointIdleDays: number; + taskDirectoryRetentionEnabled: boolean; + taskDirectoryRetentionDryRun: boolean; + taskDirectoryOrphanGraceDays: number; serveUi: boolean; uiDevMiddleware: boolean; secretsProvider: SecretProvider; @@ -352,6 +355,7 @@ export function loadConfig(): Config { 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 taskDirectoryOrphanGraceDays = Math.max(1, nonNegativeIntegerEnv("PAPERCLIP_TASK_DIRECTORY_ORPHAN_GRACE_DAYS", 7)); const bindValidationErrors = validateConfiguredBindMode({ deploymentMode, deploymentExposure, @@ -420,6 +424,9 @@ export function loadConfig(): Config { gitRestorePointsEnabled: process.env.PAPERCLIP_GIT_RESTORE_POINTS_ENABLED === "true", gitRestorePointTerminalDelayDays, gitRestorePointIdleDays, + taskDirectoryRetentionEnabled: process.env.PAPERCLIP_TASK_DIRECTORY_RETENTION_ENABLED === "true", + taskDirectoryRetentionDryRun: process.env.PAPERCLIP_TASK_DIRECTORY_RETENTION_DRY_RUN !== "false", + taskDirectoryOrphanGraceDays, serveUi: process.env.SERVE_UI !== undefined ? process.env.SERVE_UI === "true" diff --git a/server/src/index.ts b/server/src/index.ts index 905605d2e65..584e9e67697 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1481,6 +1481,11 @@ async function startServerWithDatabaseTeardown( idleMs: config.gitRestorePointIdleDays > 0 ? config.gitRestorePointIdleDays * 24 * 60 * 60 * 1000 : null, keyDurable: config.storageProvider !== "s3" || Boolean(process.env.PAPERCLIP_SECRETS_MASTER_KEY?.trim()), } : undefined, + taskDirectoryRetention: config.taskDirectoryRetentionEnabled ? { + dryRun: config.taskDirectoryRetentionDryRun, + terminalDelayMs: config.gitRestorePointTerminalDelayDays * 24 * 60 * 60 * 1000, + orphanGraceMs: config.taskDirectoryOrphanGraceDays * 24 * 60 * 60 * 1000, + } : undefined, workspaceReaperCooldownDays: config.workspaceReaperCooldownDays, workspaceCompactionDelayMs: config.workspaceCompactionDelayHours === null ? null diff --git a/server/src/services/cancelled-native-startup.ts b/server/src/services/cancelled-native-startup.ts index 53a17527f48..810c4d5939d 100644 --- a/server/src/services/cancelled-native-startup.ts +++ b/server/src/services/cancelled-native-startup.ts @@ -7,7 +7,12 @@ import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; import { canContinueCancelledRun } from "./run-cancellation.js"; import { legacyControllerBootId } from "./legacy-controller-lease.js"; -type Run = typeof heartbeatRuns.$inferSelect; +type Run = Pick; type Coordinator = typeof nativeRunFinalizations.$inferSelect; /** Candidate for explicit user Retry only, never automatic replay after Stop. */ diff --git a/server/src/services/execution-blocker-read.test.ts b/server/src/services/execution-blocker-read.test.ts new file mode 100644 index 00000000000..309385a25cb --- /dev/null +++ b/server/src/services/execution-blocker-read.test.ts @@ -0,0 +1,115 @@ +import { randomBytes, randomUUID } from "node:crypto"; +import { agents, companies, createDb, heartbeatRuns, issueRecoveryActions, issues, type Db } from "@paperclipai/db"; +import { eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; +import { getExecutionBlocker } from "./execution-blocker.js"; + +const support = await getEmbeddedPostgresTestSupport(); +describe.skipIf(!support.supported)("execution blocker PostgreSQL read projection", () => { + let database: Awaited>; + let db: ReturnType; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("execution-blocker-read-"); + db = createDb(database.connectionString); + }, 60_000); + afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); + + async function fixture(patch: Partial = {}, extraEvidence: Record = {}) { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Blocker fixture", issuePrefix: companyId.slice(0, 8) }); + await db.insert(agents).values({ id: agentId, companyId, name: "Blocker agent", adapterType: "process", status: "active" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Blocker task", status: "blocked", assigneeAgentId: agentId }); + const payload = randomBytes(128 * 1024).toString("base64"); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "failed", runtimeMode: "legacy", + startedAt: new Date("2020-01-01T00:00:00Z"), finishedAt: new Date("2020-01-01T00:01:00Z"), + errorCode: "adapter_failed", error: "Fixture failure", contextSnapshot: { issueId, prompt: payload }, + runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner", adapterConfig: { instructions: payload } }, instructions: payload }, + stdoutExcerpt: payload, ...patch, + }).returning(); + await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issueId, kind: "active_run_watchdog", + cause: "legacy_execution_requires_reconciliation", fingerprint: run.id, status: "active", nextAction: "Inspect the source run.", + evidence: { runId: run.id, ...extraEvidence } }); + return { companyId, agentId, issueId, run }; + } + + function recordRunReads(fullRows = false) { + const reads: unknown[][] = []; + const recorded = new Proxy(db, { get(target, key) { + if (key !== "select") return Reflect.get(target, key); + return (...fields: Parameters) => { + const builder = target.select(...fields), from = builder.from.bind(builder); + builder.from = ((...args: Parameters) => { + const query = args[0] === heartbeatRuns && fullRows ? target.select().from(heartbeatRuns) : from(...args); + if (args[0] === heartbeatRuns) { + const execute = query.execute.bind(query); + query.execute = async (...executeArgs: Parameters) => { + const rows = await execute(...executeArgs); reads.push(rows); return rows; + }; + } + return query; + }) as typeof builder.from; + return builder; + }; + }}); + return { recorded, reads }; + } + + async function compare(f: Awaited>) { + const baseline = recordRunReads(true), projected = recordRunReads(); + const expected = await getExecutionBlocker(baseline.recorded, f.companyId, f.issueId); + const result = await getExecutionBlocker(projected.recorded, f.companyId, f.issueId); + expect(expected).not.toBeNull(); + expect(result).toEqual(expected); + expect(projected.reads).toHaveLength(baseline.reads.length); + expect(projected.reads.flat()).toHaveLength(1); + expect(Buffer.byteLength(JSON.stringify(projected.reads))).toBeLessThan(4096); + expect(Buffer.byteLength(JSON.stringify(baseline.reads))).toBeGreaterThan(128 * 1024); + return result; + } + + it("preserves blocker and retry decisions without returning prompts, profiles, or output", async () => { + const f = await fixture(); + expect(await compare(f)).toMatchObject({ runId: f.run.id, agentId: f.agentId, canRetry: true, canContinue: false }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)))[0]).toEqual(f.run); + expect(await getExecutionBlocker(db, randomUUID(), f.issueId)).toBeNull(); + }); + + it("retains the claimed adapter and startup receipts for a cancelled pre-provider run", async () => { + const f = await fixture({ status: "cancelled", startedAt: null, errorCode: "operator_interrupted", + resultJson: { startupCancellation: { beforeNativeSelection: true }, startupPreparationSettledAt: "2020-01-01T00:01:00Z" } }); + expect(await compare(f)).toMatchObject({ canRetry: true, canContinue: false }); + await db.update(heartbeatRuns).set({ processPid: 424242 }).where(eq(heartbeatRuns.id, f.run.id)); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ canRetry: false }); + }); + + it.each(["valid", "number", "missing"])("preserves typed pre-dispatch interaction evidence (%s)", async kind => { + const context = { source: "connection_intent.resolved", connectionIntentResolution: "existing_connection", + interactionKind: "connection_intent", interactionStatus: "expired", mutation: "interaction", + ...(kind === "missing" ? {} : { interactionId: kind === "valid" ? randomUUID() : 42 }) }; + const f = await fixture({ status: "cancelled", errorCode: "issue_waiting_for_response", contextSnapshot: context, + resultJson: { stopReason: "issue_waiting_for_response", timeoutSource: "stale_queued_run_gate", + preDispatchResponseWait: { version: 1, providerWorkStarted: false } } }); + expect(await compare(f)).toMatchObject({ canRetry: kind === "valid" }); + }); + + it.each([true, false])("preserves native no-workspace evidence and cwd presence (%s)", async hasCwd => { + const f = await fixture({ runtimeMode: "native", driverKind: "grok_bot_mcp" }, { + workspaceRestoreFailure: "restore_unsafe_archive", + workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [] }, + }); + const input = { schema: "paperclip.native-execution-input.v6", + binding: { runId: f.run.id, companyId: f.companyId, agentId: f.agentId, issueId: f.issueId }, + provider: { kind: "grok_bot" }, workspace: { access: "none", ...(hasCwd ? { cwd: null } : {}) } }; + await db.update(heartbeatRuns).set({ nativeIssueId: f.issueId, + runnerProfileJson: { ...f.run.runnerProfileJson, nativeExecutionInput: input } }).where(eq(heartbeatRuns.id, f.run.id)); + expect(await compare(f)).toMatchObject({ workspaceRepairRequired: !hasCwd, canRetry: false }); + const [stored] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)); + expect(stored.runnerProfileJson?.nativeExecutionInput).toEqual(input); + }); + + it.each([null, [], "scalar", 42, { issueId: 42, taskId: false }].map(context => ({ context })))("preserves decisions for malformed context $context", async ({ context }) => { + const f = await fixture({ contextSnapshot: context as Record | null }); + expect(await compare(f)).toMatchObject({ canRetry: true }); + }); +}); diff --git a/server/src/services/execution-blocker.ts b/server/src/services/execution-blocker.ts index b1277e53f1e..4777d5947b3 100644 --- a/server/src/services/execution-blocker.ts +++ b/server/src/services/execution-blocker.ts @@ -14,6 +14,52 @@ import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; import { readNativeWorkspaceSyncReference } from "./native-runtime/native-workspace-sync.js"; import { hasVerifiedWorkspaceRepair, workspaceRecoveryDisplay } from "./workspace-recovery-receipt.js"; +const blockerContextKeys = ["issueId", "taskId", "source", "connectionIntentResolution", + "interactionKind", "interactionStatus", "mutation", "interactionId"] as const; +const blockerProfileKeys = ["adapterDispatch", "nativeExecutionInput", "nativeWorkspaceSync"] as const; +const blockerContextColumn = sql | null>`case + when ${heartbeatRuns.contextSnapshot} is null then null else ( + select coalesce(jsonb_object_agg(field.key, field.value), '{}'::jsonb) + from jsonb_each(case when jsonb_typeof(${heartbeatRuns.contextSnapshot}) = 'object' + then ${heartbeatRuns.contextSnapshot} else '{}'::jsonb end) as field(key, value) + where field.key in (${sql.join(blockerContextKeys.map(key => sql`${key}`), sql`, `)}) + ) end`; +const blockerProfileColumn = sql | null>`case + when ${heartbeatRuns.runnerProfileJson} is null then null else ( + select coalesce(jsonb_object_agg(field.key, case field.key + when 'adapterDispatch' then jsonb_build_object('adapterType', field.value->'adapterType') + else field.value end), '{}'::jsonb) + from jsonb_each(case when jsonb_typeof(${heartbeatRuns.runnerProfileJson}) = 'object' + then ${heartbeatRuns.runnerProfileJson} else '{}'::jsonb end) as field(key, value) + where field.key in (${sql.join(blockerProfileKeys.map(key => sql`${key}`), sql`, `)}) + ) end`; +const blockerRunFields = { + id: heartbeatRuns.id, + companyId: heartbeatRuns.companyId, + agentId: heartbeatRuns.agentId, + status: heartbeatRuns.status, + startedAt: heartbeatRuns.startedAt, + finishedAt: heartbeatRuns.finishedAt, + error: heartbeatRuns.error, + errorCode: heartbeatRuns.errorCode, + issueId: heartbeatRuns.issueId, + nativeIssueId: heartbeatRuns.nativeIssueId, + nativeSessionId: heartbeatRuns.nativeSessionId, + runtimeMode: heartbeatRuns.runtimeMode, + runtimeModeResolvedAt: heartbeatRuns.runtimeModeResolvedAt, + driverKind: heartbeatRuns.driverKind, + processPid: heartbeatRuns.processPid, + processGroupId: heartbeatRuns.processGroupId, + processStartedAt: heartbeatRuns.processStartedAt, + sessionIdAfter: heartbeatRuns.sessionIdAfter, + controllerBootId: heartbeatRuns.controllerBootId, + controllerLeaseExpiresAt: heartbeatRuns.controllerLeaseExpiresAt, + executionStage: heartbeatRuns.executionStage, + resultJson: heartbeatRuns.resultJson, + contextSnapshot: blockerContextColumn, + runnerProfileJson: blockerProfileColumn, +}; + export async function hasPendingRunCleanup(db: Db, companyId: string, runId: string): Promise { const [state] = await db.execute<{ pending: boolean }>(sql`SELECT ${heartbeatRuns.runnerProfileJson}->>'nativeWorkspaceFinalizationOwner' IS NOT NULL @@ -99,7 +145,7 @@ export async function getExecutionBlocker(db: Db, companyId: string, issueId: st if (!action) return null; const parsedRunId = z.string().guid().safeParse(action.evidence.runId ?? action.evidence.sourceRunId); const runId = parsedRunId.success ? parsedRunId.data : null; - const [run] = runId ? await db.select().from(heartbeatRuns).where(and( + const [run] = runId ? await db.select(blockerRunFields).from(heartbeatRuns).where(and( eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, runId), )).limit(1) : []; const withoutWorkspace = Boolean(run?.driverKind === "grok_bot_mcp" && nativeRunHasNoWorkspaceAccess(run)); diff --git a/server/src/services/execution-workspaces.ts b/server/src/services/execution-workspaces.ts index caf9a90511c..63800f66b9e 100644 --- a/server/src/services/execution-workspaces.ts +++ b/server/src/services/execution-workspaces.ts @@ -6,10 +6,15 @@ 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, gitWorkspaceLastActivityAt, hasPendingGitExpiration, listIssuesUsingGitWorkspace } from "./git-workspace-expiration.js"; +import { expireIsolatedTaskDirectory, hasPendingTaskDirectoryRetention, isIsolatedTaskDirectoryWorkspace, isRetryableTaskDirectorySkip, + mayBeIsolatedTaskDirectoryWorkspace, readTaskDirectoryRetention, restoreIsolatedTaskDirectory, supersedeTaskDirectoryRestorePoint, + sweepIsolatedTaskDirectories, inspectIsolatedTaskDirectoryFolders, taskDirectoryFoldersForHints, taskDirectoryCandidateFilter, TASK_DIRECTORY_RESTORE_POINT_EXPIRED_WARNING, TASK_DIRECTORY_RETENTION_KEY, + TASK_DIRECTORY_RETRY_MS } from "./isolated-task-directory-retention.js"; import { hasPendingIsolatedRepositoryRehomeCleanup, isIsolatedRepositoryRehomeCandidate, rehomeExpiredIsolatedRestoration, rehomeIsolatedRepositoryWorkspace, sweepOrphanedIsolatedRepositorySources } from "./isolated-repository-rehome.js"; import { hasPendingNestedCloneMigration, isNestedCloneMigrationCandidate, nextNestedCloneMigrationBatch, pendingNestedCloneMigrationNames, listNestedCloneMigrationNames, migrateNestedRepositoryClone, NESTED_CLONE_MIGRATION_RETRY_REASONS, type NestedCloneMigrationRotation } from "./nested-clone-migration.js"; import type { StorageService } from "../storage/types.js"; +import { getStorageService } from "../storage/index.js"; import { readGitCacheRestoration, restoreGitCacheBase, recoverGitCacheExpirationLocks, GitCacheRestorationFailure, gitRestorePointExpired } from "./git-workspace-restoration.js"; export const GIT_RESTORE_POINT_EXPIRED_WARNING = "The saved state of this workspace expired, so it was reopened as a fresh workspace on its branch. Unpushed and uncommitted work from before it was cleaned up is no longer available."; @@ -181,6 +186,11 @@ export function bumpExecutionWorkspaceLifecycleGeneration( }; } +function reopenableWorkspaceMode() { + return or(eq(executionWorkspaces.mode, "isolated_workspace"), + sql`(${executionWorkspaces.metadata}->${TASK_DIRECTORY_RETENTION_KEY}) IS NOT NULL`); +} + function isClosedExecutionWorkspaceStatus(status: string | null | undefined): boolean { return status === "archived" || status === "cleanup_failed"; } @@ -309,6 +319,7 @@ export type ExecutionWorkspaceServiceOptions = { nestedCloneMigrationEnabled?: boolean; nestedCloneMigrationDryRun?: boolean; gitRestorePoints?: { storage: StorageService; terminalDelayMs: number; keyDurable: boolean; idleMs?: number | null }; + taskDirectoryRetention?: { dryRun: boolean; terminalDelayMs: number; orphanGraceMs: number; perSweep?: number; trackedLimit?: number }; terminalWorkspaceInspectionBackoffMs?: number; resolvePullRequestDetails?: PullRequestMergeDetailsResolver; now?: () => Date; @@ -1760,6 +1771,80 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic let isolatedOrphanSweepDueAt = 0; let isolatedOrphanCursor: string | null = null; let isolatedOrphanKept = false; + const taskDirectoryDeadlines = new Map(); + let taskDirectoryWalkDueAt = 0; + let taskDirectoryWalkCursor: string | null = null; + let taskDirectoryWalkRetryAt: number | null = null; + const taskDirectoryFolderDue = new Map(); + const taskDirectoryHints = new Set(); + const taskDirectoryRowSkips = new Map(); + const TASK_DIRECTORY_FOLDERS_PER_SWEEP = 8; + const TASK_DIRECTORY_HINTS_PER_SWEEP = 64; + const remember = (map: Map, key: K, value: V) => { + if (!map.has(key) && map.size >= 4096) map.delete(map.keys().next().value!); + map.set(key, value); + }; + const TASK_DIRECTORY_TRACKED_LIMIT = opts.taskDirectoryRetention?.trackedLimit ?? 4096; + const rememberFolder = (key: string, dueAt: number) => { + if (!taskDirectoryFolderDue.has(key) && taskDirectoryFolderDue.size >= TASK_DIRECTORY_TRACKED_LIMIT) { + taskDirectoryWalkDueAt = Math.min(taskDirectoryWalkDueAt, dueAt); + return; + } + taskDirectoryFolderDue.set(key, dueAt); + }; + async function runTaskDirectoryFolderWork(signal?: AbortSignal) { + const settings = opts.taskDirectoryRetention!; + const options = { dryRun: settings.dryRun, now: now(), orphanGraceMs: settings.orphanGraceMs, terminalDelayMs: settings.terminalDelayMs, + signal, beforeDelete: opts.beforeTerminalWorkspaceCleanup, + restorePoint: opts.gitRestorePoints ? { storage: opts.gitRestorePoints.storage, keyDurable: opts.gitRestorePoints.keyDurable } : undefined }; + const nowMs = options.now.getTime(); + let results: Awaited> = []; + try { + if (taskDirectoryWalkDueAt <= nowMs) { + const batch = await sweepIsolatedTaskDirectories(db, { ...options, after: taskDirectoryWalkCursor, limit: TASK_DIRECTORY_FOLDERS_PER_SWEEP }); + results = batch.results; + if (batch.retryAt !== null) taskDirectoryWalkRetryAt = Math.min(taskDirectoryWalkRetryAt ?? Infinity, batch.retryAt); + taskDirectoryWalkCursor = batch.next; + if (taskDirectoryWalkCursor) taskDirectoryWalkDueAt = nowMs; + else { + taskDirectoryWalkDueAt = taskDirectoryWalkRetryAt ?? Number.POSITIVE_INFINITY; + taskDirectoryWalkRetryAt = null; + } + } else { + const hints = [...taskDirectoryHints].slice(0, TASK_DIRECTORY_HINTS_PER_SWEEP); + for (const hint of hints) taskDirectoryHints.delete(hint); + const hinted = hints.length === 0 ? [] : await taskDirectoryFoldersForHints(db, { + workspaceIds: hints.filter(hint => hint.startsWith("workspace:")).map(hint => hint.slice("workspace:".length)), + issueIds: hints.filter(hint => hint.startsWith("issue:")).map(hint => hint.slice("issue:".length)), + }); + const folders = new Map(hinted.map(folder => [`${folder.companyId}/${folder.issueId}`, folder])); + for (const [key, dueAt] of taskDirectoryFolderDue) { + if (dueAt > nowMs) continue; + const [companyId, issueId] = key.split("/"); + folders.set(key, { companyId: companyId!, issueId: issueId! }); + } + const batch = [...folders.entries()].slice(0, TASK_DIRECTORY_FOLDERS_PER_SWEEP); + for (const [key] of batch) taskDirectoryFolderDue.delete(key); + for (const [key] of [...folders.entries()].slice(TASK_DIRECTORY_FOLDERS_PER_SWEEP)) rememberFolder(key, nowMs); + results = await inspectIsolatedTaskDirectoryFolders(db, batch.map(([, folder]) => folder), options); + } + } catch (error) { + signal?.throwIfAborted(); + logger.warn({ err: error }, "isolated task directory sweep failed; it retries later"); + taskDirectoryWalkDueAt = Math.min(taskDirectoryWalkDueAt, nowMs + TASK_DIRECTORY_RETRY_MS); + } + for (const folder of results) { + const key = `${folder.companyId}/${folder.issueId}`; + if (folder.walkRetryAt !== undefined) taskDirectoryWalkDueAt = Math.min(taskDirectoryWalkDueAt, folder.walkRetryAt); + if (folder.retryAt !== undefined) rememberFolder(key, folder.retryAt); + else taskDirectoryFolderDue.delete(key); + logger.info({ event: "execution_workspace.task_directory_cleanup", dryRun: settings.dryRun, ...folder }, "Isolated task directory inspected"); + } + taskDirectoryDeadlines.clear(); + if (Number.isFinite(taskDirectoryWalkDueAt)) taskDirectoryDeadlines.set("walk", taskDirectoryWalkDueAt); + if (taskDirectoryHints.size > 0) taskDirectoryDeadlines.set("hints", now().getTime()); + if (taskDirectoryFolderDue.size > 0) taskDirectoryDeadlines.set("folders", Math.min(...taskDirectoryFolderDue.values())); + } let terminalSweepRescanRequested = false; // The reaper freezes an upper bound on updatedAt at the start of each // rotation. The scan only reads candidates at or below the bound, so a steady @@ -3258,6 +3343,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic terminalSweepInProgress = true; try { if (options.compact !== false) await purgeDeferredCompactionTrash(signal); + if (opts.taskDirectoryRetention) await runTaskDirectoryFolderWork(signal); const baseCandidateFilter = and( sql`${executionWorkspaces.sourceIssueId} IS NOT NULL`, or(and(inArray(executionWorkspaces.status, ["active", "idle", "in_review"]), isNull(executionWorkspaces.closedAt)), @@ -3269,7 +3355,8 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic sql`(${executionWorkspaces.metadata}->'nestedCloneMigrationPending') IS NOT NULL`) : undefined, opts.gitCacheExpirationEnabled && opts.isolatedSourceRehomeEnabled ? and(inArray(executionWorkspaces.status, ["archived", "cleanup_failed"]), sql`(${executionWorkspaces.metadata}->'isolatedRepositoryRehome') IS NOT NULL`, - sql`(${executionWorkspaces.metadata}->'isolatedRepositoryRehome'->>'cleanedAt') IS NULL`) : undefined), + sql`(${executionWorkspaces.metadata}->'isolatedRepositoryRehome'->>'cleanedAt') IS NULL`) : undefined, + opts.taskDirectoryRetention ? taskDirectoryCandidateFilter() : undefined), ); // Continue the scan after the previous sweep's last row. The keyset // predicate uses the same (updatedAt, id) order as the query, so each @@ -3354,6 +3441,34 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic compactionFailed: 0, }; const readinessByWorkingTree = new Map>(); + let taskDirectoryBudget = opts.taskDirectoryRetention?.perSweep ?? 8; + const expireTaskDirectory = async (workspace: ExecutionWorkspaceRow, notBefore: Date) => { + const settings = opts.taskDirectoryRetention!; + const skipped = taskDirectoryRowSkips.get(workspace.id); + if (skipped && skipped.updatedAt === workspace.updatedAt.getTime() && (skipped.retryAt === null || skipped.retryAt > now().getTime())) { + if (skipped.retryAt !== null) terminalWorkspaceDeadlines.set(workspace.id, skipped.retryAt); + return; + } + if (taskDirectoryBudget <= 0) { + terminalWorkspaceDeadlines.set(workspace.id, now().getTime()); + return; + } + taskDirectoryBudget -= 1; + const expiration = await expireIsolatedTaskDirectory(db, workspace, { dryRun: settings.dryRun, now: now(), notBefore, signal, + beforeDelete: opts.beforeTerminalWorkspaceCleanup, + restorePoint: opts.gitRestorePoints ? { storage: opts.gitRestorePoints.storage, keyDurable: opts.gitRestorePoints.keyDurable } : undefined }); + if (expiration.candidate) result.eligible += 1; + if (expiration.expired) result.archived += 1; + const retryAt = !expiration.expired && isRetryableTaskDirectorySkip(expiration.skipReason) ? now().getTime() + TASK_DIRECTORY_RETRY_MS : null; + if (retryAt !== null) terminalWorkspaceDeadlines.set(workspace.id, retryAt); + else terminalWorkspaceDeadlines.delete(workspace.id); + if (expiration.expired) taskDirectoryRowSkips.delete(workspace.id); + else remember(taskDirectoryRowSkips, workspace.id, { updatedAt: workspace.updatedAt.getTime(), retryAt, issueId: workspace.sourceIssueId }); + logger.info({ event: "execution_workspace.task_directory_retention", executionWorkspaceId: workspace.id, + companyId: workspace.companyId, sourceIssueId: workspace.sourceIssueId, dryRun: settings.dryRun, + outcome: expiration.expired ? "removed" : settings.dryRun && expiration.candidate ? "dry_run" : "skipped", ...expiration }, + "Isolated task directory retention inspected"); + }; for (let workspace of candidates) { signal?.throwIfAborted(); @@ -3428,6 +3543,10 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic } if (nestedOnly) continue; } + if (opts.taskDirectoryRetention && hasPendingTaskDirectoryRetention(workspace)) { + await expireTaskDirectory(workspace, now()); + continue; + } const issueTree = await listWorkspaceIssueTree(workspace); const sourceIssue = issueTree.find((issue) => issue.id === workspace.sourceIssueId); const reopenPending = metadataHasReopenPendingConsumption( @@ -3482,6 +3601,15 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic } const expirationTerminalAt = Math.max(...expirationTree.map(issue => issueTerminalTimestamp(issue)!.getTime())); const latestTerminalAt = treeOpen ? expirationTerminalAt : Math.max(...issueTree.map(issue => issueTerminalTimestamp(issue)!.getTime())); + if (opts.taskDirectoryRetention && !treeOpen && !reopenPending && !isClosedExecutionWorkspaceStatus(workspace.status) + && mayBeIsolatedTaskDirectoryWorkspace(workspace) && await isIsolatedTaskDirectoryWorkspace(workspace)) { + const dueAt = latestTerminalAt + opts.taskDirectoryRetention.terminalDelayMs; + if (dueAt > now().getTime()) { + terminalWorkspaceDeadlines.set(workspace.id, dueAt); + result.skippedCooldown += 1; + } else await expireTaskDirectory(workspace, new Date(dueAt)); + continue; + } // Compaction runs once per terminal cycle, after its own delay and // before the archive cooldown, so a workspace the reaper cannot archive // yet (or ever) stops holding regenerable files. @@ -3862,6 +3990,17 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic terminalWorkspaceWorkSettled: (hints?: ReadonlySet) => { if (terminalSweepInProgress || terminalSweepCursor !== null) terminalSweepRescanRequested = true; pullRequestStateCache.clear(); + if (opts.taskDirectoryRetention) { + if (!hints) taskDirectoryRowSkips.clear(); + for (const hint of hints ?? []) { + if (!hint.startsWith("workspace:") && !hint.startsWith("issue:")) continue; + if (taskDirectoryHints.size < TASK_DIRECTORY_TRACKED_LIMIT) taskDirectoryHints.add(hint); + else taskDirectoryWalkDueAt = Math.min(taskDirectoryWalkDueAt, now().getTime()); + taskDirectoryDeadlines.set("hints", now().getTime()); + if (hint.startsWith("workspace:")) taskDirectoryRowSkips.delete(hint.slice("workspace:".length)); + else for (const [id, skip] of taskDirectoryRowSkips) if (`issue:${skip.issueId}` === hint) taskDirectoryRowSkips.delete(id); + } + } if (!hints) { terminalInspections.clear(); return; } for (const [id, inspection] of terminalInspections) { if (hints.has(`workspace:${id}`) || [...inspection.issueIds].some(issueId => hints.has(`issue:${issueId}`))) { @@ -3871,8 +4010,10 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic }, nextTerminalWorkspaceSweepAt: async () => { - if (terminalWorkspaceDeadlines.size === 0 && isolatedRehomeDeadlines.size === 0 && nestedCloneMigrationDeadlines.size === 0) return null; - const next = [...terminalWorkspaceDeadlines.values(), ...isolatedRehomeDeadlines.values(), ...nestedCloneMigrationDeadlines.values()].reduce((earliest, deadline) => Math.min(earliest, deadline), Infinity); + if (terminalWorkspaceDeadlines.size === 0 && isolatedRehomeDeadlines.size === 0 && nestedCloneMigrationDeadlines.size === 0 + && taskDirectoryDeadlines.size === 0) return null; + const next = [...terminalWorkspaceDeadlines.values(), ...isolatedRehomeDeadlines.values(), ...nestedCloneMigrationDeadlines.values(), + ...taskDirectoryDeadlines.values()].reduce((earliest, deadline) => Math.min(earliest, deadline), Infinity); const nowMs = now().getTime(); if (next <= nowMs && opts.terminalWorkspaceInspectionBackoffMs) { return Math.min(nowMs + opts.terminalWorkspaceInspectionBackoffMs, @@ -3931,7 +4072,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic )); const authorizedSnapshot = boundSnapshot ? await db.select().from(executionWorkspaces).where(and( eq(executionWorkspaces.id, input.workspaceId), eq(executionWorkspaces.companyId, issue.companyId), - eq(executionWorkspaces.mode, "isolated_workspace"), + reopenableWorkspaceMode(), )).then(rows => rows[0] ?? null) : null; if (!authorizedSnapshot) return { ok: false, code: "not_reopenable", message: "Execution workspace is not reopenable" }; await assertTaskWorkspaceExecutionAccess(db, input.authorizationActor, issue.companyId, { issueId: issue.id, workspaceId: authorizedSnapshot.id }); @@ -3939,7 +4080,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic const reopen = async (): Promise => { const freshSnapshot = await db.select().from(executionWorkspaces).where(and( eq(executionWorkspaces.id, input.workspaceId), eq(executionWorkspaces.companyId, issue.companyId), - eq(executionWorkspaces.mode, "isolated_workspace"), + reopenableWorkspaceMode(), )).then(rows => rows[0] ?? null); if (!freshSnapshot || freshSnapshot.sourceIssueId !== sourceIssueId) { return { ok: false, code: "not_reopenable", message: "Execution workspace source changed during reopen" }; @@ -4068,7 +4209,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic .where(and( eq(executionWorkspaces.id, input.workspaceId), eq(executionWorkspaces.companyId, issue.companyId), - eq(executionWorkspaces.mode, "isolated_workspace"), + reopenableWorkspaceMode(), )) .then((rows) => rows[0] ?? null); if (!row) { @@ -4148,7 +4289,15 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic }); let rebuildError: string | null = null; + let taskDirectoryWarning: string | null = null; + const taskDirectoryState = readTaskDirectoryRetention(row); try { + if (taskDirectoryState) { + const restored = await restoreIsolatedTaskDirectory({ storage: opts.gitRestorePoints?.storage ?? getStorageService(), workspace: row }); + if (restored.expired) taskDirectoryWarning = TASK_DIRECTORY_RESTORE_POINT_EXPIRED_WARNING; + logger.info({ event: "execution_workspace.task_directory_restore", executionWorkspaceId: row.id, companyId: row.companyId, + issueId: issue.id, ...restored }, "Isolated task directory restored for reopen"); + } const [assignee, organization, sourceProject] = await Promise.all([ boundTask.assigneeAgentId ? tx.select({ companyId: agents.companyId, permissions: agents.permissions }).from(agents) .where(and(eq(agents.id, boundTask.assigneeAgentId), eq(agents.companyId, row.companyId))) @@ -4324,6 +4473,10 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic // receive newer task commits; the next expiration records a fresh recipe. if (freshStart) delete nextMetadata.gitCacheRestoration; else if (recipe) nextMetadata.gitCacheRestoration = { ...nextMetadata.gitCacheRestoration as Record, restoredAt: now().toISOString() }; + if (taskDirectoryState) { + delete nextMetadata[TASK_DIRECTORY_RETENTION_KEY]; + await supersedeTaskDirectoryRestorePoint(tx, row); + } const activeMetadata = setMetadataReopenPendingConsumption(nextMetadata, now()); await measureRunPhase("workspace.notification", () => notifyTerminalWorkspaceWork(tx, `workspace:${row.id}`)); const activeRow = await tx @@ -4369,14 +4522,16 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic outcome: "reopened", generation: nextGeneration, ...(freshStart ? { restorePointExpired: true, warning: GIT_RESTORE_POINT_EXPIRED_WARNING } : {}), + ...(taskDirectoryWarning ? { restorePointExpired: true, warning: taskDirectoryWarning } : {}), }, }); - if (freshStart) { + const reopenWarning = freshStart ? GIT_RESTORE_POINT_EXPIRED_WARNING : taskDirectoryWarning; + if (reopenWarning) { const { issueService } = await import("./issues.js"); - await issueService(db).addComment(issue.id, GIT_RESTORE_POINT_EXPIRED_WARNING, {}, { authorType: "system" }, tx); + await issueService(db).addComment(issue.id, reopenWarning, {}, { authorType: "system" }, tx); } return { ok: true, reopened: true, workspace: toExecutionWorkspace(activeRow), generation: nextGeneration, - ...(freshStart ? { warning: GIT_RESTORE_POINT_EXPIRED_WARNING } : {}) }; + ...(reopenWarning ? { warning: reopenWarning } : {}) }; }).catch((error): ReopenClosedIsolatedExecutionWorkspaceResult => { logger.warn({ error, executionWorkspaceId: snapshot.id }, "Execution workspace reopen failed"); return { ok: false, code: "rebuild_failed", message: recipe ? new GitCacheRestorationFailure().message : "Failed to rebuild the execution workspace" }; diff --git a/server/src/services/git-restore-point-ledger.ts b/server/src/services/git-restore-point-ledger.ts index 2070e2f1b1e..04e7c6333f7 100644 --- a/server/src/services/git-restore-point-ledger.ts +++ b/server/src/services/git-restore-point-ledger.ts @@ -77,7 +77,9 @@ export function gitRestorePointRetentionService(db: Db, options: GitRestorePoint eq(executionWorkspaceRestorePoints.status, "active"), eq(executionWorkspaceRestorePoints.reviewGeneration, row.reviewGeneration), ); - const deleteAfter = !tree.exists || tree.open || !tree.closedAt ? null : new Date(tree.closedAt.getTime() + retentionMs); + const retainedFrom = tree.closedAt && row.activatedAt && row.objectKey.includes("/task-directory/") + ? Math.max(tree.closedAt.getTime(), row.activatedAt.getTime()) : tree.closedAt?.getTime(); + const deleteAfter = !tree.exists || tree.open || retainedFrom === undefined ? null : new Date(retainedFrom + retentionMs); if (!tree.exists || (deleteAfter && deleteAfter.getTime() <= at.getTime())) { const claimed = await db.update(executionWorkspaceRestorePoints) .set({ status: tree.exists ? "expiring" : "delete_requested", lastError: null, updatedAt: at }) diff --git a/server/src/services/git-restore-point-storage.ts b/server/src/services/git-restore-point-storage.ts index 87bcae36578..bb32379d92a 100644 --- a/server/src/services/git-restore-point-storage.ts +++ b/server/src/services/git-restore-point-storage.ts @@ -18,6 +18,7 @@ import { } from "./git-restore-point.js"; import { notifyGitRestorePointWork } from "./git-restore-point-ledger.js"; import { gitCacheCommand } from "./git-workspace-restoration.js"; +import { TASK_DIRECTORY_RETENTION_KEY } from "./isolated-task-directory-snapshot.js"; export const GIT_RESTORE_POINT_SCHEMA = "paperclip.git-restore-point.v1"; const NAMESPACE = "git-restore-points"; @@ -99,45 +100,11 @@ export async function storeGitRestorePoint(input: { const identity = { companyId: input.workspace.companyId, workspaceId: input.workspace.id, restorePointId: id, generation: input.generation }; const sealed = await sealRestorePoint(captured.payload, identity); const objectKey = restorePointObjectKey(input.workspace.companyId, input.workspace.id, input.generation, id, input.checkout); - const [ledger] = await input.db.transaction(async (tx) => { - await notifyGitRestorePointWork(tx, `workspace:${input.workspace.id}`); - await tx.update(executionWorkspaceRestorePoints).set({ status: "superseded", updatedAt: new Date() }) - .where(and(eq(executionWorkspaceRestorePoints.executionWorkspaceId, input.workspace.id), - eq(executionWorkspaceRestorePoints.generation, input.generation), - eq(executionWorkspaceRestorePoints.checkoutName, input.checkout ?? ""), - eq(executionWorkspaceRestorePoints.status, "pending"))); - return tx.insert(executionWorkspaceRestorePoints).values({ - companyId: input.workspace.companyId, - executionWorkspaceId: input.workspace.id, - sourceIssueId: input.workspace.sourceIssueId, - generation: input.generation, - checkoutName: input.checkout ?? "", - objectKey, - sha256: sealed.sha256, - byteSize: sealed.ciphertext.length, - status: "pending", - }).returning({ id: executionWorkspaceRestorePoints.id }); - }); - try { - await input.storage.putFile({ - objectKey, - companyId: input.workspace.companyId, - namespace: NAMESPACE, - originalFilename: null, - contentType: "application/octet-stream", - body: sealed.ciphertext, - }); - const stored = await readObject(input.storage, input.workspace.companyId, objectKey, input.signal); - if (createHash("sha256").update(stored).digest("hex") !== sealed.sha256) { - throw new RestorePointRestoreFailure("stored_restore_point_mismatch"); - } - } catch (error) { - await abandonGitRestorePoint(input.db, ledger!.id).catch(() => undefined); - throw error; - } + const ledgerId = await storeSealedRestorePoint({ db: input.db, storage: input.storage, workspace: input.workspace, + generation: input.generation, checkoutName: input.checkout ?? "", objectKey, sealed, signal: input.signal }); const { manifest } = captured; return { - ledgerId: ledger!.id, + ledgerId, manifest, ref: { schema: GIT_RESTORE_POINT_SCHEMA, @@ -159,6 +126,55 @@ export async function storeGitRestorePoint(input: { }; } +export async function storeSealedRestorePoint(input: { + db: Db; + storage: StorageService; + workspace: { companyId: string; id: string; sourceIssueId: string | null }; + generation: number; + checkoutName: string; + objectKey: string; + sealed: { ciphertext: Buffer; sha256: string }; + signal?: AbortSignal; +}): Promise { + const [ledger] = await input.db.transaction(async (tx) => { + await notifyGitRestorePointWork(tx, `workspace:${input.workspace.id}`); + await tx.update(executionWorkspaceRestorePoints).set({ status: "superseded", updatedAt: new Date() }) + .where(and(eq(executionWorkspaceRestorePoints.executionWorkspaceId, input.workspace.id), + eq(executionWorkspaceRestorePoints.generation, input.generation), + eq(executionWorkspaceRestorePoints.checkoutName, input.checkoutName), + eq(executionWorkspaceRestorePoints.status, "pending"))); + return tx.insert(executionWorkspaceRestorePoints).values({ + companyId: input.workspace.companyId, + executionWorkspaceId: input.workspace.id, + sourceIssueId: input.workspace.sourceIssueId, + generation: input.generation, + checkoutName: input.checkoutName, + objectKey: input.objectKey, + sha256: input.sealed.sha256, + byteSize: input.sealed.ciphertext.length, + status: "pending", + }).returning({ id: executionWorkspaceRestorePoints.id }); + }); + try { + await input.storage.putFile({ + objectKey: input.objectKey, + companyId: input.workspace.companyId, + namespace: NAMESPACE, + originalFilename: null, + contentType: "application/octet-stream", + body: input.sealed.ciphertext, + }); + const stored = await readRestorePointObject(input.storage, input.workspace.companyId, input.objectKey, input.signal); + if (createHash("sha256").update(stored).digest("hex") !== input.sealed.sha256) { + throw new RestorePointRestoreFailure("stored_restore_point_mismatch"); + } + } catch (error) { + await abandonGitRestorePoint(input.db, ledger!.id).catch(() => undefined); + throw error; + } + return ledger!.id; +} + export async function activateGitRestorePoints(tx: Tx, input: { ledgerIds: string[]; workspaceId: string; at: Date }) { if (input.ledgerIds.length === 0) return; await notifyGitRestorePointWork(tx, `workspace:${input.workspaceId}`); @@ -200,7 +216,7 @@ export async function abandonGitRestorePoint(db: Db, ledgerId: string) { .where(and(eq(executionWorkspaceRestorePoints.id, ledgerId), inArray(executionWorkspaceRestorePoints.status, ["pending", "active"]))); } -async function readObject(storage: StorageService, companyId: string, objectKey: string, signal?: AbortSignal) { +export async function readRestorePointObject(storage: StorageService, companyId: string, objectKey: string, signal?: AbortSignal) { const { stream } = await storage.getObject(companyId, objectKey); const chunks: Buffer[] = []; let total = 0; @@ -221,7 +237,7 @@ export async function loadGitRestorePoint(input: { signal?: AbortSignal; }): Promise<{ payload: Buffer; manifest: RestorePointManifest }> { if (input.ref.expiredAt) throw new RestorePointRestoreFailure("restore_point_expired"); - const ciphertext = await readObject(input.storage, input.workspace.companyId, input.ref.objectKey, input.signal) + const ciphertext = await readRestorePointObject(input.storage, input.workspace.companyId, input.ref.objectKey, input.signal) .catch((error) => { if (error instanceof RestorePointRestoreFailure) throw error; throw new RestorePointRestoreFailure("restore_point_unavailable"); @@ -278,28 +294,36 @@ export async function applyGitRestorePointToWorktree(input: { if (!matches) await applyRestorePoint({ cwd: input.worktreePath, manifest, signal: input.signal }); } +function expireRestorePointRecipe(key: string, recipe: Record, objectKey: string, at: string) { + const expire = (ref: Record | undefined) => ref?.objectKey === objectKey ? { ...ref, expiredAt: at } : ref; + if (key === TASK_DIRECTORY_RETENTION_KEY) return recipe.restorePoint ? { ...recipe, restorePoint: expire(recipe.restorePoint) } : recipe; + return { + ...recipe, + ...(recipe.restorePoint ? { restorePoint: expire(recipe.restorePoint) } : {}), + ...(recipe.contents ? { contents: { ...recipe.contents, checkouts: (recipe.contents.checkouts ?? []).map((checkout: Record) => + checkout.restorePoint ? { ...checkout, restorePoint: expire(checkout.restorePoint) } : checkout) } } : {}), + }; +} + export async function markGitRestorePointExpired(db: Db, restorePoint: { companyId: string; executionWorkspaceId: string; objectKey: string }) { const at = new Date().toISOString(); - for (let attempt = 0; attempt < 3; attempt += 1) { - const [row] = await db.select({ metadata: executionWorkspaces.metadata }).from(executionWorkspaces) - .where(and(eq(executionWorkspaces.id, restorePoint.executionWorkspaceId), eq(executionWorkspaces.companyId, restorePoint.companyId))); - const recipe = row?.metadata?.gitCacheRestoration as Record | undefined; - if (!recipe) return; - const expire = (ref: Record | undefined) => ref?.objectKey === restorePoint.objectKey ? { ...ref, expiredAt: at } : ref; - const next = { - ...recipe, - ...(recipe.restorePoint ? { restorePoint: expire(recipe.restorePoint) } : {}), - ...(recipe.contents ? { contents: { ...recipe.contents, checkouts: (recipe.contents.checkouts ?? []).map((checkout: Record) => - checkout.restorePoint ? { ...checkout, restorePoint: expire(checkout.restorePoint) } : checkout) } } : {}), - }; - if (JSON.stringify(next) === JSON.stringify(recipe)) return; - const updated = await db.update(executionWorkspaces).set({ - metadata: sql`jsonb_set(${executionWorkspaces.metadata}, '{gitCacheRestoration}', ${JSON.stringify(next)}::jsonb)`, - updatedAt: new Date(), - }).where(and(eq(executionWorkspaces.id, restorePoint.executionWorkspaceId), eq(executionWorkspaces.companyId, restorePoint.companyId), - sql`${executionWorkspaces.metadata}->'gitCacheRestoration' = ${JSON.stringify(recipe)}::jsonb`)) - .returning({ id: executionWorkspaces.id }); - if (updated.length === 1) return; + for (const key of ["gitCacheRestoration", TASK_DIRECTORY_RETENTION_KEY]) { + let settled = false; + for (let attempt = 0; attempt < 3 && !settled; attempt += 1) { + const [row] = await db.select({ metadata: executionWorkspaces.metadata }).from(executionWorkspaces) + .where(and(eq(executionWorkspaces.id, restorePoint.executionWorkspaceId), eq(executionWorkspaces.companyId, restorePoint.companyId))); + const recipe = row?.metadata?.[key] as Record | undefined; + if (!recipe) { settled = true; break; } + const next = expireRestorePointRecipe(key, recipe, restorePoint.objectKey, at); + if (JSON.stringify(next) === JSON.stringify(recipe)) { settled = true; break; } + const updated = await db.update(executionWorkspaces).set({ + metadata: sql`jsonb_set(${executionWorkspaces.metadata}, ${`{${key}}`}::text[], ${JSON.stringify(next)}::jsonb)`, + updatedAt: new Date(), + }).where(and(eq(executionWorkspaces.id, restorePoint.executionWorkspaceId), eq(executionWorkspaces.companyId, restorePoint.companyId), + sql`${executionWorkspaces.metadata}->${key} = ${JSON.stringify(recipe)}::jsonb`)) + .returning({ id: executionWorkspaces.id }); + settled = updated.length === 1; + } + if (!settled) throw new RestorePointRestoreFailure("restore_point_expiry_conflict"); } - throw new RestorePointRestoreFailure("restore_point_expiry_conflict"); } diff --git a/server/src/services/git-workspace-expiration.ts b/server/src/services/git-workspace-expiration.ts index c551f7471a5..12cb8e52dda 100644 --- a/server/src/services/git-workspace-expiration.ts +++ b/server/src/services/git-workspace-expiration.ts @@ -8,6 +8,7 @@ import { executionWorkspaces, issues, heartbeatRuns, nativeRunFinalizations, env workspaceRuntimeServices, issueRecoveryActions, projectWorkspaces, withDedicatedDbConnection, type Db } from "@paperclipai/db"; import { createGitRemoteAuthProvider } from "./git-credentials.js"; import type { StorageService } from "../storage/types.js"; +import { TASK_DIRECTORY_RETENTION_KEY } from "./isolated-task-directory-snapshot.js"; import { currentRestorePointSnapshot, RestorePointSkip, type RestorePointLimits } from "./git-restore-point.js"; import { abandonGitRestorePoint, abandonGitRestorePointsByKey, activateGitRestorePoints, restorePointHistoryUnchanged, restorePointSnapshotMatches, storeGitRestorePoint } from "./git-restore-point-storage.js"; import { withIsolatedRepositorySourceLock, verifiedRecordedIsolatedRemoteSource } from "./isolated-remote-workspace-source.js"; @@ -151,8 +152,13 @@ export async function gitWorkspaceLastActivityAt(db: Db | Tx, workspace: Workspa return at ? new Date(at) : null; } -export async function isGitWorkspaceExpirationEligible(tx: Tx, workspace: Workspace, now: Date, idleCutoff?: Date) { +export async function isGitWorkspaceExpirationEligible(tx: Tx, workspace: Workspace, now: Date, idleCutoff?: Date, + options: { closedSiblingsAllowed?: boolean } = {}) { if (workspace.metadata?.reopenPendingConsumption === true) return false; + const siblingBlocks = options.closedSiblingsAllowed + ? sql`AND (other.status NOT IN ('archived', 'cleanup_failed') OR other.closed_at IS NULL + OR other.metadata->${TASK_DIRECTORY_RETENTION_KEY}->>'removed' = 'false')` + : sql``; 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) @@ -172,7 +178,7 @@ export async function isGitWorkspaceExpirationEligible(tx: Tx, workspace: Worksp 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 left(other.cwd, length(${workspace.cwd}) + 1) = ${workspace.cwd} || '/') ${siblingBlocks}) OR EXISTS (SELECT 1 FROM ${projectWorkspaces} WHERE cwd = ${workspace.cwd} OR left(cwd, length(${workspace.cwd}) + 1) = ${workspace.cwd} || '/') THEN false ELSE ( diff --git a/server/src/services/heartbeat/environment-selection.ts b/server/src/services/heartbeat/environment-selection.ts index 959a352902a..9910ab0b5b7 100644 --- a/server/src/services/heartbeat/environment-selection.ts +++ b/server/src/services/heartbeat/environment-selection.ts @@ -18,6 +18,7 @@ import { } from "../execution-workspace-policy.js"; import { assertTaskWorkspaceExecutionAccess, assertTaskWorkspaceSourceProjectAccess } from "../task-workspace-access.js"; import { shouldUseIsolatedTaskDirectory } from "../isolated-task-directory.js"; +import { readTaskDirectoryRetention } from "../isolated-task-directory-retention.js"; import { findNativeChatWorkspaceScope, nativeChatWorkspaceCwd, @@ -116,7 +117,8 @@ export async function selectHeartbeatEnvironment(db: Db, input: HeartbeatEnviron if (existingExecutionWorkspace && issueId && !isDotRun && !fresh && existingExecutionWorkspace.id === readNonEmptyString(issueRef?.executionWorkspaceId) && (existingExecutionWorkspace.status === "archived" || existingExecutionWorkspace.status === "cleanup_failed") - && readGitCacheExpiration(existingExecutionWorkspace as Parameters[0])?.workspaceRemoved) { + && (readGitCacheExpiration(existingExecutionWorkspace as Parameters[0])?.workspaceRemoved + || readTaskDirectoryRetention(existingExecutionWorkspace) !== null)) { 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"); diff --git a/server/src/services/isolated-task-directory-retention.ts b/server/src/services/isolated-task-directory-retention.ts new file mode 100644 index 00000000000..cb5a8dc8a37 --- /dev/null +++ b/server/src/services/isolated-task-directory-retention.ts @@ -0,0 +1,627 @@ +import { randomUUID } from "node:crypto"; +import fs from "node:fs/promises"; +import path from "node:path"; +import { and, eq, inArray, or, sql } from "drizzle-orm"; +import { environmentLeases, executionWorkspaceRestorePoints, executionWorkspaces, issueRecoveryActions, issues, nativeRunFinalizations, + projectWorkspaces, withDedicatedDbConnection, workspaceRuntimeServices, type Db } from "@paperclipai/db"; +import type { StoredSecretVersionMaterial } from "../secrets/types.js"; +import type { StorageService } from "../storage/types.js"; +import { resolvePaperclipInstanceRoot } from "../home-paths.js"; +import { openRestorePoint, sealRestorePoint } from "./git-restore-point.js"; +import { notifyGitRestorePointWork } from "./git-restore-point-ledger.js"; +import { abandonGitRestorePoint, activateGitRestorePoints, readRestorePointObject, storeSealedRestorePoint } from "./git-restore-point-storage.js"; +import { allocatedBytes, describeGitExpirationFailure, expirationScope, isGitWorkspaceExpirationEligible, type GitExpirationFailure } from "./git-workspace-expiration.js"; +import { IsolatedRepositorySourceBusy, withIsolatedRepositorySourceLock } from "./isolated-remote-workspace-source.js"; +import { logActivity } from "./activity-log.js"; +import { + captureTaskDirectorySnapshot, + inspectTaskDirectory, + materializeTaskDirectorySnapshot, + newestTaskDirectoryMtime, + removeTaskDirectoryTree, + shortSiblingName, + TASK_DIRECTORY_RETENTION_KEY, + TaskDirectoryRestoreFailure, + TaskDirectorySnapshotSkip, + type TaskDirectorySnapshotLimits, +} from "./isolated-task-directory-snapshot.js"; + +export { TASK_DIRECTORY_RETENTION_KEY } from "./isolated-task-directory-snapshot.js"; +export const TASK_DIRECTORY_RETENTION_SCHEMA = "paperclip.task-directory-retention.v1"; +export const TASK_DIRECTORY_RESTORE_POINT_SCHEMA = "paperclip.task-directory-restore-point.v1"; +export const TASK_DIRECTORY_RETENTION_REASON = "isolated task directory retention"; +export const TASK_DIRECTORY_RESTORE_POINT_EXPIRED_WARNING = "The saved files of this task directory expired, so it was reopened as an empty directory. Files from before it was cleaned up are no longer available."; +export const TASK_DIRECTORY_RETRY_MS = 60 * 60 * 1000; +export const TASK_DIRECTORY_RECHECK_MS = 24 * 60 * 60 * 1000; +const TRASH_GRACE_MS = 60 * 60 * 1000; +const NAMESPACE = "git-restore-points"; +const MAX_RESTORE_POINT_BYTES = 512 * 1024 * 1024; +const IDENTITY = /^[a-zA-Z0-9_-]+$/; +const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; +const TRASH = /^\.(expired|restoring)-[0-9a-f]{8,}$/; +const RETRYABLE_SKIPS = new Set(["source_busy", "lifecycle_busy", "workspace_changed", "live_or_unresolved_consumer", + "task_directory_changed", "inspection_failed"]); + +type Workspace = typeof executionWorkspaces.$inferSelect; + +export type TaskDirectoryRestorePointRef = { + schema: typeof TASK_DIRECTORY_RESTORE_POINT_SCHEMA; + id: string; + objectKey: string; + sha256: string; + byteSize: number; + key: StoredSecretVersionMaterial; + generation: number; + digest: string; + files: number; + excluded: number; + expiredAt?: string; +}; + +export type TaskDirectoryRetention = { + schema: typeof TASK_DIRECTORY_RETENTION_SCHEMA; + generation: number; + removed: boolean; + trash: string; + restorePoint: TaskDirectoryRestorePointRef; + removedAt?: string; +}; + +export type TaskDirectoryRestorePointOptions = { storage: StorageService; keyDurable: boolean; limits?: TaskDirectorySnapshotLimits }; + +export type TaskDirectoryRetentionResult = { + candidate: boolean; + expired: boolean; + restorePoint: boolean; + candidateBytes: number; + reclaimedBytes: number; + skipReason: string | null; + failure?: GitExpirationFailure; +}; + +export function readTaskDirectoryRetention(workspace: { metadata?: Record | null }): TaskDirectoryRetention | null { + const state = workspace.metadata?.[TASK_DIRECTORY_RETENTION_KEY] as TaskDirectoryRetention | undefined; + return state?.schema === TASK_DIRECTORY_RETENTION_SCHEMA ? state : null; +} + +export function hasPendingTaskDirectoryRetention(workspace: Pick) { + const state = readTaskDirectoryRetention(workspace); + return Boolean(state && !state.removed && ["archived", "cleanup_failed"].includes(workspace.status)); +} + +export function isRetryableTaskDirectorySkip(reason: string | null) { + return reason !== null && RETRYABLE_SKIPS.has(reason); +} + +async function instanceRoot() { + return fs.realpath(resolvePaperclipInstanceRoot()).catch(() => null); +} + +export async function isolatedTaskDirectoriesRoot() { + const root = await instanceRoot(); + return root ? path.join(root, "isolated-workspaces") : null; +} + +export async function canonicalTaskDirectory(companyId: string, issueId: string) { + if (!IDENTITY.test(companyId) || !IDENTITY.test(issueId)) return null; + const root = await isolatedTaskDirectoriesRoot(); + return root ? path.join(root, companyId, issueId) : null; +} + +export function mayBeIsolatedTaskDirectoryWorkspace(workspace: Pick) { + return Boolean(workspace.sourceIssueId && workspace.cwd) && workspace.strategyType === "project_primary" + && workspace.providerType === "local_fs" && !workspace.repoUrl && !workspace.branchName + && (workspace.providerRef === null || workspace.providerRef === workspace.cwd) + && (workspace.mode === "isolated_workspace" || workspace.mode === "shared_workspace"); +} + +export async function isIsolatedTaskDirectoryWorkspace(workspace: Pick) { + if (!mayBeIsolatedTaskDirectoryWorkspace(workspace)) return false; + return workspace.cwd === await canonicalTaskDirectory(workspace.companyId, workspace.sourceIssueId!); +} + +function restorePointObjectKey(companyId: string, workspaceId: string, generation: number, id: string) { + return `${companyId}/${NAMESPACE}/${workspaceId}/${generation}/task-directory/${id}.pcrp`; +} + +function validateRestorePointRef(ref: TaskDirectoryRestorePointRef | null | undefined, workspace: { companyId: string; id: string }) { + if (!ref || ref.schema !== TASK_DIRECTORY_RESTORE_POINT_SCHEMA || typeof ref.id !== "string" || !UUID.test(ref.id) + || !Number.isSafeInteger(ref.generation) || ref.generation < 1 + || ref.objectKey !== restorePointObjectKey(workspace.companyId, workspace.id, ref.generation, ref.id) + || !/^[0-9a-f]{64}$/.test(ref.sha256) || !/^[0-9a-f]{64}$/.test(ref.digest) + || !Number.isSafeInteger(ref.byteSize) || ref.byteSize <= 0 || ref.byteSize > MAX_RESTORE_POINT_BYTES + || !ref.key || typeof ref.key !== "object" || Array.isArray(ref.key)) { + throw new TaskDirectoryRestoreFailure("invalid_restore_point_reference"); + } + return ref; +} + +function skipReason(error: unknown) { + if (error instanceof TaskDirectorySnapshotSkip) return error.reason; + if (error instanceof IsolatedRepositorySourceBusy) return "source_busy"; + if (error instanceof Error && (error as NodeJS.ErrnoException).code === "ENOENT") return "task_directory_changed"; + return error instanceof Error && /^[a-z][a-z_]+$/.test(error.message) ? error.message : "inspection_failed"; +} + +function recordSkip(result: { skipReason: string | null; failure?: GitExpirationFailure }, error: unknown) { + result.skipReason = skipReason(error); + if (result.skipReason === "inspection_failed") result.failure = describeGitExpirationFailure(error); +} + +async function directoryState(target: string) { + const stat = await fs.lstat(target).catch((error: NodeJS.ErrnoException) => { + if (error.code === "ENOENT") return null; + throw error; + }); + if (!stat) return "missing" as const; + if (stat.isSymbolicLink() || !stat.isDirectory() || await fs.realpath(target) !== target) throw new Error("unsafe_task_directory"); + return "present" as const; +} + +async function exists(target: string) { + return fs.lstat(target).then(() => true, (error: NodeJS.ErrnoException) => { + if (error.code === "ENOENT") return false; + throw error; + }); +} + +async function issueBoundToWorkspace(db: Db | Parameters[0]>[0], workspace: Workspace) { + const [bound] = await db.select({ id: issues.id }).from(issues).where(and(eq(issues.companyId, workspace.companyId), + eq(issues.id, workspace.sourceIssueId!), eq(issues.executionWorkspaceId, workspace.id))); + return Boolean(bound); +} + +async function storeTaskDirectoryRestorePoint(db: Db, storage: StorageService, workspace: Workspace, generation: number, + captured: Awaited>, signal?: AbortSignal) { + const id = randomUUID(); + const sealed = await sealRestorePoint(captured.payload, { companyId: workspace.companyId, workspaceId: workspace.id, restorePointId: id, generation }); + const objectKey = restorePointObjectKey(workspace.companyId, workspace.id, generation, id); + const ledgerId = await storeSealedRestorePoint({ db, storage, workspace: { companyId: workspace.companyId, id: workspace.id, + sourceIssueId: workspace.sourceIssueId }, generation, checkoutName: "", objectKey, sealed, signal }); + const ref: TaskDirectoryRestorePointRef = { schema: TASK_DIRECTORY_RESTORE_POINT_SCHEMA, id, objectKey, sha256: sealed.sha256, + byteSize: sealed.ciphertext.length, key: sealed.key, generation, digest: captured.digest, files: captured.files, excluded: captured.excluded }; + return { ledgerId, ref }; +} + +export async function expireIsolatedTaskDirectory(db: Db, snapshot: Workspace, options: { + dryRun: boolean; + now: Date; + notBefore: Date; + restorePoint?: TaskDirectoryRestorePointOptions; + signal?: AbortSignal; + beforeDelete?: (workspace: Workspace) => Promise; +}): Promise { + const result: TaskDirectoryRetentionResult = { candidate: false, expired: false, restorePoint: false, candidateBytes: 0, reclaimedBytes: 0, skipReason: null }; + let pending: string | null = null; + if (!await isIsolatedTaskDirectoryWorkspace(snapshot)) return { ...result, skipReason: "not_task_directory" }; + const closed = ["archived", "cleanup_failed"].includes(snapshot.status); + const existing = readTaskDirectoryRetention(snapshot); + if (existing?.removed) return { ...result, skipReason: "already_removed" }; + if (!existing && options.now.getTime() < options.notBefore.getTime()) return { ...result, skipReason: "not_due" }; + if (!existing && !options.dryRun) { + if (!options.restorePoint) return { ...result, skipReason: "restore_points_disabled" }; + if (!options.restorePoint.keyDurable) return { ...result, skipReason: "restore_point_key_not_durable" }; + } + try { + return await withIsolatedRepositorySourceLock(db, snapshot.companyId, snapshot.sourceIssueId!, () => + withDedicatedDbConnection(db, connection => connection.transaction(async tx => { + const acquired = await tx.execute<{ locked: boolean }>(sql`select pg_try_advisory_xact_lock(hashtextextended(${`execution_workspace_lifecycle:${snapshot.id}`}, 0)) as locked`); + 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))); + if (!workspace || workspace.updatedAt.getTime() !== snapshot.updatedAt.getTime()) throw new Error("workspace_changed"); + if (!await issueBoundToWorkspace(tx, workspace)) throw new Error("issue_not_bound"); + if (!await isGitWorkspaceExpirationEligible(tx, workspace, options.now, undefined, { closedSiblingsAllowed: true })) { + throw new Error("live_or_unresolved_consumer"); + } + const cwd = workspace.cwd!; + let state = readTaskDirectoryRetention(workspace); + const present = await directoryState(cwd) === "present"; + if (!state) { + if (!present) throw new Error("missing_task_directory"); + result.candidateBytes = await allocatedBytes(cwd, () => { options.signal?.throwIfAborted(); return true; }); + const captured = await captureTaskDirectorySnapshot(cwd, { limits: options.restorePoint?.limits, signal: options.signal }); + result.candidate = true; + if (options.dryRun) return result; + const generation = Number(workspace.metadata?.lifecycleGeneration ?? 0) + 1; + const stored = await storeTaskDirectoryRestorePoint(db, options.restorePoint!.storage, workspace, generation, captured, options.signal); + pending = stored.ledgerId; + result.restorePoint = true; + state = { schema: TASK_DIRECTORY_RETENTION_SCHEMA, generation, removed: false, + trash: shortSiblingName(cwd, ".expired-"), restorePoint: stored.ref }; + if (await exists(state.trash)) throw new Error("unknown_task_directory_trash"); + const metadata = { ...workspace.metadata, lifecycleGeneration: generation, [TASK_DIRECTORY_RETENTION_KEY]: state }; + const archive = closed ? { metadata, updatedAt: options.now } : { status: "archived", closedAt: options.now, + cleanupEligibleAt: options.now, cleanupReason: TASK_DIRECTORY_RETENTION_REASON, metadata, updatedAt: options.now }; + [workspace] = await db.transaction(async archiveTx => { + const rows = await archiveTx.update(executionWorkspaces).set(archive).where(eq(executionWorkspaces.id, workspace!.id)).returning(); + await activateGitRestorePoints(archiveTx, { ledgerIds: [stored.ledgerId], workspaceId: workspace!.id, at: options.now }); + return rows; + }); + pending = null; + if (!closed) { + await logActivity(db, { + companyId: workspace!.companyId, + actorType: "system", + actorId: "workspace_terminality_reaper", + action: "execution_workspace.task_directory_archived", + entityType: "execution_workspace", + entityId: workspace!.id, + details: { sourceIssueId: workspace!.sourceIssueId, restorePointId: stored.ref.id, cleanupReason: TASK_DIRECTORY_RETENTION_REASON }, + }); + } + } else { + result.candidate = true; + if (present) result.candidateBytes = await allocatedBytes(cwd, () => { options.signal?.throwIfAborted(); return true; }); + if (options.dryRun) return result; + } + if (!workspace) throw new Error("workspace_changed"); + await options.beforeDelete?.(workspace); + if (!await isGitWorkspaceExpirationEligible(tx, workspace, options.now, undefined, { closedSiblingsAllowed: true })) { + throw new Error("live_or_unresolved_consumer"); + } + if (await directoryState(cwd) === "present") { + const current = await inspectTaskDirectory(cwd, { limits: options.restorePoint?.limits, signal: options.signal }) + .catch((error: unknown) => { + if (error instanceof TaskDirectorySnapshotSkip) return null; + throw error; + }); + if (!current || current.digest !== state.restorePoint.digest) { + if (workspace.cleanupReason === TASK_DIRECTORY_RETENTION_REASON) { + const metadata: Record = { ...workspace.metadata, lifecycleGeneration: state.generation + 1 }; + delete metadata[TASK_DIRECTORY_RETENTION_KEY]; + await db.update(executionWorkspaces).set({ status: "active", closedAt: null, cleanupEligibleAt: null, cleanupReason: null, + metadata, updatedAt: new Date() }).where(eq(executionWorkspaces.id, workspace.id)); + } else { + const metadata: Record = { ...workspace.metadata }; + delete metadata[TASK_DIRECTORY_RETENTION_KEY]; + await db.update(executionWorkspaces).set({ metadata, updatedAt: new Date() }).where(eq(executionWorkspaces.id, workspace.id)); + } + await abandonTaskDirectoryRestorePoint(db, workspace.id, state.restorePoint.objectKey); + throw new Error("task_directory_changed"); + } + await fs.rename(cwd, state.trash); + } + if (await exists(state.trash)) { + const bytes = await allocatedBytes(state.trash, () => { options.signal?.throwIfAborted(); return true; }).catch(() => result.candidateBytes); + await removeTaskDirectoryTree(state.trash); + result.reclaimedBytes = bytes; + } + const removedAt = new Date().toISOString(); + await db.update(executionWorkspaces).set({ + metadata: sql`jsonb_set(jsonb_set(${executionWorkspaces.metadata}, ${`{${TASK_DIRECTORY_RETENTION_KEY},removed}`}::text[], 'true'::jsonb), + ${`{${TASK_DIRECTORY_RETENTION_KEY},removedAt}`}::text[], ${JSON.stringify(removedAt)}::jsonb)`, + updatedAt: new Date(), + }).where(and(eq(executionWorkspaces.id, workspace.id), + sql`${executionWorkspaces.metadata}->${TASK_DIRECTORY_RETENTION_KEY}->'restorePoint'->>'objectKey' = ${state.restorePoint.objectKey}`)); + await logActivity(db, { + companyId: workspace.companyId, + actorType: "system", + actorId: "workspace_terminality_reaper", + action: "execution_workspace.task_directory_removed", + entityType: "execution_workspace", + entityId: workspace.id, + details: { sourceIssueId: workspace.sourceIssueId, restorePointId: state.restorePoint.id, reclaimedBytes: result.reclaimedBytes, + cleanupReason: TASK_DIRECTORY_RETENTION_REASON }, + }); + result.expired = true; + return result; + })), true); + } catch (error) { + if (pending) await abandonGitRestorePoint(db, pending).catch(() => undefined); + options.signal?.throwIfAborted(); + recordSkip(result, error); + return result; + } +} + +async function abandonTaskDirectoryRestorePoint(db: Db | Parameters[0]>[0], workspaceId: string, objectKey: string) { + await db.transaction(async tx => { + await notifyGitRestorePointWork(tx, `workspace:${workspaceId}`); + await tx.update(executionWorkspaceRestorePoints).set({ status: "superseded", updatedAt: new Date() }) + .where(and(eq(executionWorkspaceRestorePoints.executionWorkspaceId, workspaceId), eq(executionWorkspaceRestorePoints.objectKey, objectKey), + inArray(executionWorkspaceRestorePoints.status, ["pending", "active"]))); + }); +} + +export async function supersedeTaskDirectoryRestorePoint(tx: Parameters[0]>[0], workspace: { id: string; metadata?: Record | null }) { + const state = readTaskDirectoryRetention(workspace); + if (state) await abandonTaskDirectoryRestorePoint(tx, workspace.id, state.restorePoint.objectKey); +} + +export async function restoreIsolatedTaskDirectory(input: { + storage: StorageService; + workspace: Pick; + signal?: AbortSignal; +}): Promise<{ restored: boolean; expired: boolean }> { + const state = readTaskDirectoryRetention(input.workspace); + if (!state) return { restored: false, expired: false }; + if (!await isIsolatedTaskDirectoryWorkspace(input.workspace)) throw new TaskDirectoryRestoreFailure("task_directory_identity_changed"); + const cwd = input.workspace.cwd!; + const parent = path.dirname(cwd); + await fs.mkdir(parent, { recursive: true, mode: 0o700 }); + if (await fs.realpath(parent) !== parent) throw new TaskDirectoryRestoreFailure("task_directory_identity_changed"); + const ref = validateRestorePointRef(state.restorePoint, input.workspace); + if (await directoryState(cwd) === "present") { + if (!state.removed) return { restored: false, expired: false }; + if ((await fs.readdir(cwd)).length === 0) await fs.rmdir(cwd); + else { + const current = await inspectTaskDirectory(cwd, { signal: input.signal }).catch(() => null); + if (current?.digest === ref.digest) return { restored: false, expired: false }; + throw new TaskDirectoryRestoreFailure("task_directory_diverged"); + } + } + if (ref.expiredAt) { + await fs.mkdir(cwd, { mode: 0o700 }); + return { restored: false, expired: true }; + } + const ciphertext = await readRestorePointObject(input.storage, input.workspace.companyId, ref.objectKey, input.signal).catch(() => { + throw new TaskDirectoryRestoreFailure("restore_point_unavailable"); + }); + if (ciphertext.length !== ref.byteSize) throw new TaskDirectoryRestoreFailure("ciphertext_size_mismatch"); + const payload = await openRestorePoint({ ciphertext, sha256: ref.sha256, key: ref.key, + identity: { companyId: input.workspace.companyId, workspaceId: input.workspace.id, restorePointId: ref.id, generation: ref.generation } }); + const staging = await materializeTaskDirectorySnapshot({ payload, target: cwd, expectedDigest: ref.digest, signal: input.signal }); + try { + await fs.rename(staging, cwd); + } catch (error) { + await removeTaskDirectoryTree(staging).catch(() => undefined); + throw error; + } + return { restored: true, expired: false }; +} + +export type TaskDirectoryFolderResult = { + companyId: string; + issueId: string; + outcome: "removed" | "preserved_and_removed" | "dry_run" | "skipped"; + reason: string | null; + candidateBytes: number; + reclaimedBytes: number; + executionWorkspaceId?: string; + retryAt?: number; + walkRetryAt?: number; + failure?: GitExpirationFailure; +}; + +type IssueTree = { exists: boolean; open: boolean; closedAt: Date | null }; + +async function issueTreeState(db: Db, companyId: string, issueId: string): Promise { + const rows = await db.execute<{ total: number; open: number; closed_at: string | Date | null }>(sql` + WITH RECURSIVE tree AS ( + SELECT id, status, completed_at, cancelled_at, updated_at FROM ${issues} + WHERE company_id = ${companyId} AND id = ${issueId} + 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 = ${companyId} + ) + SELECT count(*)::int AS total, + count(*) FILTER (WHERE status NOT IN ('done', 'cancelled'))::int AS open, + max(CASE WHEN status = 'done' THEN coalesce(completed_at, updated_at) + WHEN status = 'cancelled' THEN coalesce(cancelled_at, updated_at) END) AS closed_at + FROM tree`); + const state = rows[0]; + if (!state || state.total === 0) return { exists: false, open: false, closedAt: null }; + return { exists: true, open: state.open > 0, closedAt: state.closed_at ? new Date(state.closed_at) : null }; +} + +async function folderRows(db: Db, folder: string) { + return db.select().from(executionWorkspaces).where(or(eq(executionWorkspaces.cwd, folder), eq(executionWorkspaces.providerRef, folder), + sql`left(${executionWorkspaces.cwd}, length(${folder}) + 1) = ${folder} || '/'`, + sql`left(${executionWorkspaces.providerRef}, length(${folder}) + 1) = ${folder} || '/'`)); +} + +async function orphanConsumer(db: Db, companyId: string, issueId: string, folder: string) { + const scope = { companyId, sourceIssueId: issueId, id: "00000000-0000-0000-0000-000000000000", cwd: folder } as Workspace; + const rows = await db.execute<{ reason: string | null }>(sql`${expirationScope(scope)} + SELECT CASE + WHEN EXISTS (SELECT 1 FROM ${projectWorkspaces} WHERE cwd = ${folder} OR left(cwd, length(${folder}) + 1) = ${folder} || '/') THEN 'project_workspace' + WHEN EXISTS (SELECT 1 FROM ${workspaceRuntimeServices} WHERE status <> 'stopped' + AND (cwd = ${folder} OR left(cwd, length(${folder}) + 1) = ${folder} || '/')) THEN 'live_runtime_service' + WHEN 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()) THEN 'live_run' + WHEN EXISTS (SELECT 1 FROM ${nativeRunFinalizations} f WHERE f.company_id = ${companyId} + AND (f.issue_id IN (SELECT id FROM tree) OR f.run_id IN (SELECT id FROM relevant_runs)) + AND (f.lease_owner IS NOT NULL OR f.phase NOT IN ('committed', 'terminal_failure'))) THEN 'unsettled_finalization' + WHEN EXISTS (SELECT 1 FROM ${environmentLeases} l WHERE l.company_id = ${companyId} + AND (l.heartbeat_run_id IN (SELECT id FROM relevant_runs) OR l.issue_id IN (SELECT id FROM tree) + OR l.execution_workspace_id IN (SELECT w.id FROM ${executionWorkspaces} w WHERE w.cwd = ${folder} OR w.provider_ref = ${folder} + OR left(w.cwd, length(${folder}) + 1) = ${folder} || '/' OR left(w.provider_ref, length(${folder}) + 1) = ${folder} || '/')) + AND (l.status = 'pending_cleanup' OR l.cleanup_status = 'failed' OR l.metadata ? 'nativeWorkspaceExportResume' + OR (l.released_at IS NULL AND NOT coalesce(l.status = 'retained' AND l.lease_policy = 'reuse_by_environment' AND l.cleanup_status = 'success', false)))) THEN 'live_lease' + WHEN EXISTS (SELECT 1 FROM ${issueRecoveryActions} WHERE company_id = ${companyId} + AND source_issue_id IN (SELECT id FROM tree) AND status IN ('active', 'escalated')) THEN 'recovery_action' + ELSE NULL END AS reason`); + return rows[0]?.reason ?? null; +} + +async function listDirectory(target: string) { + const stat = await fs.lstat(target).catch(() => null); + if (!stat || !stat.isDirectory() || stat.isSymbolicLink()) return []; + return (await fs.readdir(target)).sort(); +} + +type FolderSweepOptions = { + dryRun: boolean; + now: Date; + orphanGraceMs: number; + terminalDelayMs: number; + restorePoint?: TaskDirectoryRestorePointOptions; + signal?: AbortSignal; + beforeDelete?: (workspace: Workspace) => Promise; +}; + +const CONSUMER_SKIPS = new Set(["live_workspace", "retention_pending", "project_workspace", "live_runtime_service", "live_run", + "unsettled_finalization", "live_lease", "recovery_action"]); + +async function inspectTaskDirectoryFolder(db: Db, companyId: string, name: string, options: FolderSweepOptions): Promise { + const nowMs = options.now.getTime(); + const root = await isolatedTaskDirectoriesRoot(); + if (!root) return null; + const companyDir = path.join(root, companyId); + const folder = path.join(companyDir, name); + const result: TaskDirectoryFolderResult = { companyId, issueId: name, outcome: "skipped", reason: null, candidateBytes: 0, reclaimedBytes: 0 }; + const retry = (at: number) => { result.retryAt = result.retryAt === undefined ? at : Math.min(result.retryAt, at); }; + try { + if (!UUID.test(name) || !UUID.test(companyId)) throw new Error("unrecognized_task_directory_name"); + const stat = await fs.lstat(folder).catch((error: NodeJS.ErrnoException) => { + if (error.code === "ENOENT") return null; + throw error; + }); + if (!stat) return null; + if (stat.isSymbolicLink() || !stat.isDirectory()) throw new Error("unsafe_task_directory"); + const rows = await folderRows(db, folder); + if (rows.some(row => !["archived", "cleanup_failed"].includes(row.status) || row.closedAt === null)) return null; + if (rows.some(row => hasPendingTaskDirectoryRetention(row))) throw new Error("retention_pending"); + const [issue] = await db.select({ id: issues.id, executionWorkspaceId: issues.executionWorkspaceId }).from(issues) + .where(and(eq(issues.companyId, companyId), eq(issues.id, name))); + const tree = issue ? await issueTreeState(db, companyId, name) : { exists: false, open: false, closedAt: null }; + if (tree.exists && (tree.open || !tree.closedAt)) throw new Error("issue_open"); + const bound = issue ? rows.find(row => row.id === issue.executionWorkspaceId) : undefined; + if (bound) { + if (bound.cwd !== folder || !await isIsolatedTaskDirectoryWorkspace(bound)) throw new Error("bound_workspace"); + result.executionWorkspaceId = bound.id; + const dueAt = tree.closedAt!.getTime() + options.terminalDelayMs; + if (dueAt > nowMs) { retry(dueAt); throw new Error("not_due"); } + const expiration = await expireIsolatedTaskDirectory(db, bound, { dryRun: options.dryRun, now: options.now, + notBefore: new Date(dueAt), restorePoint: options.restorePoint, signal: options.signal, beforeDelete: options.beforeDelete }); + result.candidateBytes = expiration.candidateBytes; + result.reclaimedBytes = expiration.reclaimedBytes; + result.reason = expiration.skipReason; + if (expiration.failure) result.failure = expiration.failure; + result.outcome = expiration.expired ? "preserved_and_removed" : options.dryRun && expiration.candidate ? "dry_run" : "skipped"; + if (isRetryableTaskDirectorySkip(expiration.skipReason)) retry(nowMs + TASK_DIRECTORY_RETRY_MS); + return result; + } + if (tree.exists && tree.closedAt!.getTime() + options.orphanGraceMs > nowMs) { + retry(tree.closedAt!.getTime() + options.orphanGraceMs); + throw new Error("grace_pending"); + } + const newest = await newestTaskDirectoryMtime(folder, options.signal); + if (newest + options.orphanGraceMs > nowMs) { + retry(newest + options.orphanGraceMs); + throw new Error("recently_modified"); + } + await withIsolatedRepositorySourceLock(db, companyId, name, async () => { + if (await directoryState(folder) !== "present") throw new Error("task_directory_changed"); + const current = await folderRows(db, folder); + if (current.some(row => !["archived", "cleanup_failed"].includes(row.status) || row.closedAt === null)) throw new Error("live_workspace"); + if (current.some(row => hasPendingTaskDirectoryRetention(row))) throw new Error("retention_pending"); + const [currentIssue] = await db.select({ executionWorkspaceId: issues.executionWorkspaceId }).from(issues) + .where(and(eq(issues.companyId, companyId), eq(issues.id, name))); + if (Boolean(currentIssue) !== Boolean(issue) || (currentIssue && currentIssue.executionWorkspaceId !== issue!.executionWorkspaceId)) { + throw new Error("task_directory_changed"); + } + const consumer = await orphanConsumer(db, companyId, name, folder); + if (consumer) throw new Error(consumer); + if (tree.exists) { + const latest = await issueTreeState(db, companyId, name); + if (latest.open || !latest.closedAt || latest.closedAt.getTime() + options.orphanGraceMs > nowMs) throw new Error("issue_open"); + } + result.candidateBytes = await allocatedBytes(folder, () => { options.signal?.throwIfAborted(); return true; }).catch(() => 0); + if (options.dryRun) { + result.outcome = "dry_run"; + return; + } + const trash = shortSiblingName(folder, ".expired-"); + await fs.rename(folder, trash); + await removeTaskDirectoryTree(trash).catch(() => { + result.walkRetryAt = nowMs + TASK_DIRECTORY_RETRY_MS; + throw new Error("task_directory_trash_pending"); + }); + result.outcome = "removed"; + result.reclaimedBytes = result.candidateBytes; + }, true); + } catch (error) { + options.signal?.throwIfAborted(); + result.reason = skipReason(error); + if (result.reason === "inspection_failed") result.failure = describeGitExpirationFailure(error); + if (isRetryableTaskDirectorySkip(result.reason) || CONSUMER_SKIPS.has(result.reason)) retry(nowMs + TASK_DIRECTORY_RETRY_MS); + } + return result; +} + +export async function inspectIsolatedTaskDirectoryFolders(db: Db, folders: Array<{ companyId: string; issueId: string }>, options: FolderSweepOptions) { + const results: TaskDirectoryFolderResult[] = []; + for (const folder of folders) { + options.signal?.throwIfAborted(); + if (!IDENTITY.test(folder.companyId) || !IDENTITY.test(folder.issueId)) continue; + const result = await inspectTaskDirectoryFolder(db, folder.companyId, folder.issueId, options); + if (result) results.push(result); + } + return results; +} + +export async function sweepIsolatedTaskDirectories(db: Db, options: FolderSweepOptions & { after?: string | null; limit?: number }): + Promise<{ results: TaskDirectoryFolderResult[]; next: string | null; retryAt: number | null }> { + const results: TaskDirectoryFolderResult[] = []; + const limit = options.limit ?? 8; + const nowMs = options.now.getTime(); + let retryAt: number | null = null; + const retry = (at: number) => { retryAt = retryAt === null ? at : Math.min(retryAt, at); }; + const root = await isolatedTaskDirectoriesRoot(); + if (!root) return { results, next: null, retryAt }; + const rootStat = await fs.lstat(root).catch(() => null); + if (!rootStat || !rootStat.isDirectory() || rootStat.isSymbolicLink()) return { results, next: null, retryAt }; + const [afterCompany, afterIssue] = options.after?.split("/") ?? []; + let inspected = 0; + let lastPosition: string | null = null; + for (const companyId of await listDirectory(root)) { + if (!IDENTITY.test(companyId)) continue; + const companyDir = path.join(root, companyId); + for (const name of await listDirectory(companyDir)) { + options.signal?.throwIfAborted(); + if (afterCompany !== undefined && (companyId < afterCompany || (companyId === afterCompany && name <= afterIssue!))) continue; + if (inspected >= limit) return { results, next: lastPosition, retryAt }; + inspected += 1; + lastPosition = `${companyId}/${name}`; + if (TRASH.test(name)) { + const folder = path.join(companyDir, name); + const stat = await fs.lstat(folder).catch(() => null); + if (!stat || options.dryRun) continue; + const touched = Math.max(stat.mtimeMs, stat.ctimeMs); + if (touched + TRASH_GRACE_MS > nowMs) { retry(touched + TRASH_GRACE_MS); continue; } + await removeTaskDirectoryTree(folder).catch(() => retry(nowMs + TASK_DIRECTORY_RETRY_MS)); + continue; + } + const result = await inspectTaskDirectoryFolder(db, companyId, name, options); + if (result) results.push(result); + } + } + return { results, next: null, retryAt }; +} + +export async function taskDirectoryFoldersForHints(db: Db, hints: { workspaceIds: string[]; issueIds: string[] }) { + const folders = new Map(); + const workspaceIds = hints.workspaceIds.filter(id => UUID.test(id)); + const issueIds = hints.issueIds.filter(id => UUID.test(id)); + if (workspaceIds.length > 0) { + const rows = await db.select({ companyId: executionWorkspaces.companyId, sourceIssueId: executionWorkspaces.sourceIssueId }) + .from(executionWorkspaces).where(and(inArray(executionWorkspaces.id, workspaceIds), + inArray(executionWorkspaces.status, ["archived", "cleanup_failed"]), eq(executionWorkspaces.strategyType, "project_primary"))); + for (const row of rows) if (row.sourceIssueId) folders.set(`${row.companyId}/${row.sourceIssueId}`, { companyId: row.companyId, issueId: row.sourceIssueId }); + } + if (issueIds.length > 0) { + const rows = await db.execute<{ id: string; company_id: string }>(sql` + WITH RECURSIVE closed_up AS ( + SELECT id, parent_id, company_id, 0 AS depth FROM ${issues} + WHERE id IN (${sql.join(issueIds.map(id => sql`${id}::uuid`), sql`, `)}) AND status IN ('done', 'cancelled') + UNION + SELECT p.id, p.parent_id, p.company_id, c.depth + 1 FROM ${issues} p + JOIN closed_up c ON p.id = c.parent_id AND p.company_id = c.company_id WHERE c.depth < 64 + ) + SELECT DISTINCT id, company_id FROM closed_up`); + for (const row of rows) folders.set(`${row.company_id}/${row.id}`, { companyId: row.company_id, issueId: row.id }); + } + const present: Array<{ companyId: string; issueId: string }> = []; + for (const folder of folders.values()) { + const target = await canonicalTaskDirectory(folder.companyId, folder.issueId); + if (target && await exists(target)) present.push(folder); + } + return present; +} + +export function taskDirectoryCandidateFilter() { + return and(inArray(executionWorkspaces.status, ["archived", "cleanup_failed"]), + sql`${executionWorkspaces.metadata}->${TASK_DIRECTORY_RETENTION_KEY}->>'removed' = 'false'`); +} diff --git a/server/src/services/isolated-task-directory-snapshot.ts b/server/src/services/isolated-task-directory-snapshot.ts new file mode 100644 index 00000000000..1eafdfb126f --- /dev/null +++ b/server/src/services/isolated-task-directory-snapshot.ts @@ -0,0 +1,265 @@ +import { createHash, randomBytes } from "node:crypto"; +import fs from "node:fs/promises"; +import path from "node:path"; +import { DEFAULT_RESTORE_POINT_LIMITS } from "./git-restore-point.js"; + +export const TASK_DIRECTORY_RETENTION_KEY = "isolatedTaskDirectoryRetention"; +export const TASK_DIRECTORY_SNAPSHOT_SCHEMA = "paperclip.task-directory-snapshot.v1"; +const PAYLOAD_MAGIC = Buffer.from("PCTD-PLAIN1\n"); +const MAX_MANIFEST_BYTES = 64 * 1024 * 1024; + +export type TaskDirectorySnapshotLimits = { maxFiles: number; maxFileBytes: number; maxTotalBytes: number }; + +export const DEFAULT_TASK_DIRECTORY_SNAPSHOT_LIMITS: TaskDirectorySnapshotLimits = { + maxFiles: DEFAULT_RESTORE_POINT_LIMITS.maxFiles, + maxFileBytes: DEFAULT_RESTORE_POINT_LIMITS.maxFileBytes, + maxTotalBytes: DEFAULT_RESTORE_POINT_LIMITS.maxTotalBytes, +}; + +export type TaskDirectoryEntry = + | { path: string; type: "directory"; mode: number } + | { path: string; type: "file"; mode: number; size: number; sha256: string } + | { path: string; type: "symlink"; target: string }; + +export type TaskDirectorySnapshotManifest = { + schema: typeof TASK_DIRECTORY_SNAPSHOT_SCHEMA; + entries: TaskDirectoryEntry[]; + excluded: string[]; +}; + +export class TaskDirectorySnapshotSkip extends Error { + constructor(readonly reason: string) { + super(reason); + this.name = "TaskDirectorySnapshotSkip"; + } +} + +export class TaskDirectoryRestoreFailure extends Error { + constructor(readonly reason: string) { + super(`The task directory could not be restored: ${reason}`); + this.name = "TaskDirectoryRestoreFailure"; + } +} + +const bySortKey = (a: string, b: string) => (a < b ? -1 : a > b ? 1 : 0); + +type WalkOptions = { + limits: TaskDirectorySnapshotLimits; + signal?: AbortSignal; + readFile: (absolute: string, relative: string, size: number) => Promise; +}; + +async function walk(root: string, options: WalkOptions) { + const rootStat = await fs.lstat(root); + if (!rootStat.isDirectory() || rootStat.isSymbolicLink()) throw new TaskDirectorySnapshotSkip("unsafe_task_directory"); + const entries: TaskDirectoryEntry[] = []; + const excluded: string[] = []; + let files = 0; + let total = 0; + let newestMtimeMs = rootStat.mtimeMs; + const visit = async (relativeDir: string) => { + options.signal?.throwIfAborted(); + const absoluteDir = relativeDir ? path.join(root, relativeDir) : root; + const names = (await fs.readdir(absoluteDir)).sort(bySortKey); + const hasPackageManifest = names.includes("package.json") + && (await fs.lstat(path.join(absoluteDir, "package.json"))).isFile(); + for (const name of names) { + options.signal?.throwIfAborted(); + const relative = relativeDir ? `${relativeDir}/${name}` : name; + const absolute = path.join(absoluteDir, name); + const stat = await fs.lstat(absolute); + newestMtimeMs = Math.max(newestMtimeMs, stat.mtimeMs); + if (stat.isSymbolicLink()) { + entries.push({ path: relative, type: "symlink", target: await fs.readlink(absolute) }); + continue; + } + if (stat.isDirectory()) { + if (name === "node_modules" && hasPackageManifest) { + excluded.push(relative); + continue; + } + entries.push({ path: relative, type: "directory", mode: stat.mode & 0o777 }); + await visit(relative); + continue; + } + if (!stat.isFile()) throw new TaskDirectorySnapshotSkip("unsupported_filesystem_entry"); + files += 1; + total += stat.size; + if (files > options.limits.maxFiles) throw new TaskDirectorySnapshotSkip("task_directory_too_many_files"); + if (stat.size > options.limits.maxFileBytes) throw new TaskDirectorySnapshotSkip("task_directory_file_too_large"); + if (total > options.limits.maxTotalBytes) throw new TaskDirectorySnapshotSkip("task_directory_too_large"); + entries.push({ path: relative, type: "file", mode: stat.mode & 0o777, size: stat.size, + sha256: await options.readFile(absolute, relative, stat.size) }); + } + }; + await visit(""); + return { entries, excluded, newestMtimeMs }; +} + +export function taskDirectoryManifestDigest(manifest: Pick) { + return createHash("sha256").update(JSON.stringify([TASK_DIRECTORY_SNAPSHOT_SCHEMA, manifest.entries])).digest("hex"); +} + +async function hashFile(absolute: string, size: number) { + const handle = await fs.open(absolute, fs.constants.O_RDONLY | fs.constants.O_NOFOLLOW); + try { + const hash = createHash("sha256"); + let read = 0; + for await (const chunk of handle.createReadStream({ autoClose: false })) { + read += (chunk as Buffer).length; + hash.update(chunk as Buffer); + } + if (read !== size) throw new TaskDirectorySnapshotSkip("task_directory_changed"); + return hash.digest("hex"); + } finally { + await handle.close(); + } +} + +export async function inspectTaskDirectory(root: string, options: { limits?: TaskDirectorySnapshotLimits; signal?: AbortSignal } = {}) { + const walked = await walk(root, { limits: options.limits ?? DEFAULT_TASK_DIRECTORY_SNAPSHOT_LIMITS, signal: options.signal, + readFile: (absolute, _relative, size) => hashFile(absolute, size) }); + return { digest: taskDirectoryManifestDigest(walked), entries: walked.entries.length, excluded: walked.excluded.length, + newestMtimeMs: walked.newestMtimeMs }; +} + +export async function newestTaskDirectoryMtime(root: string, signal?: AbortSignal): Promise { + const stat = await fs.lstat(root); + let newest = stat.mtimeMs; + if (!stat.isDirectory() || stat.isSymbolicLink()) return newest; + for (const name of await fs.readdir(root)) { + signal?.throwIfAborted(); + newest = Math.max(newest, await newestTaskDirectoryMtime(path.join(root, name), signal)); + } + return newest; +} + +export async function captureTaskDirectorySnapshot(root: string, options: { limits?: TaskDirectorySnapshotLimits; signal?: AbortSignal } = {}) { + const bodies: Buffer[] = []; + const walked = await walk(root, { limits: options.limits ?? DEFAULT_TASK_DIRECTORY_SNAPSHOT_LIMITS, signal: options.signal, + readFile: async (absolute, _relative, size) => { + const handle = await fs.open(absolute, fs.constants.O_RDONLY | fs.constants.O_NOFOLLOW); + try { + const body = await handle.readFile(); + if (body.length !== size) throw new TaskDirectorySnapshotSkip("task_directory_changed"); + bodies.push(body); + return createHash("sha256").update(body).digest("hex"); + } finally { + await handle.close(); + } + } }); + const manifest: TaskDirectorySnapshotManifest = { schema: TASK_DIRECTORY_SNAPSHOT_SCHEMA, entries: walked.entries, excluded: walked.excluded }; + const encoded = Buffer.from(JSON.stringify(manifest)); + const length = Buffer.alloc(4); + length.writeUInt32BE(encoded.length); + const payload = Buffer.concat([PAYLOAD_MAGIC, length, encoded, ...bodies]); + try { decodeTaskDirectorySnapshot(payload); } + catch { throw new TaskDirectorySnapshotSkip("task_directory_not_restorable"); } + return { + payload, + digest: taskDirectoryManifestDigest(manifest), + files: walked.entries.filter(entry => entry.type === "file").length, + excluded: walked.excluded.length, + }; +} + +function validRelativePath(value: unknown): value is string { + if (typeof value !== "string" || value.length === 0 || value.length > 4096 || value.includes("\0")) return false; + if (path.sep !== "/" && value.includes(path.sep)) return false; + if (value.startsWith("/")) return false; + return value.split("/").every(part => part.length > 0 && part !== "." && part !== ".."); +} + +export function decodeTaskDirectorySnapshot(payload: Buffer): { manifest: TaskDirectorySnapshotManifest; bodies: Buffer } { + if (payload.length < PAYLOAD_MAGIC.length + 4 || !payload.subarray(0, PAYLOAD_MAGIC.length).equals(PAYLOAD_MAGIC)) { + throw new TaskDirectoryRestoreFailure("invalid_snapshot_payload"); + } + const length = payload.readUInt32BE(PAYLOAD_MAGIC.length); + const start = PAYLOAD_MAGIC.length + 4; + if (length > MAX_MANIFEST_BYTES || start + length > payload.length) throw new TaskDirectoryRestoreFailure("invalid_snapshot_payload"); + let manifest: TaskDirectorySnapshotManifest; + try { manifest = JSON.parse(payload.subarray(start, start + length).toString("utf8")); } + catch { throw new TaskDirectoryRestoreFailure("invalid_snapshot_payload"); } + if (!manifest || manifest.schema !== TASK_DIRECTORY_SNAPSHOT_SCHEMA || !Array.isArray(manifest.entries) || !Array.isArray(manifest.excluded)) { + throw new TaskDirectoryRestoreFailure("invalid_snapshot_payload"); + } + const kinds = new Map(); + let bodyBytes = 0; + for (const entry of manifest.entries) { + if (!entry || !validRelativePath(entry.path) || kinds.has(entry.path)) throw new TaskDirectoryRestoreFailure("invalid_snapshot_entry"); + const parent = entry.path.includes("/") ? entry.path.slice(0, entry.path.lastIndexOf("/")) : null; + if (parent !== null && kinds.get(parent) !== "directory") throw new TaskDirectoryRestoreFailure("invalid_snapshot_entry"); + if (entry.type === "file") { + if (!Number.isSafeInteger(entry.size) || entry.size < 0 || !/^[0-9a-f]{64}$/.test(entry.sha256) + || !Number.isInteger(entry.mode) || entry.mode < 0 || entry.mode > 0o777) throw new TaskDirectoryRestoreFailure("invalid_snapshot_entry"); + bodyBytes += entry.size; + } else if (entry.type === "directory") { + if (!Number.isInteger(entry.mode) || entry.mode < 0 || entry.mode > 0o777) throw new TaskDirectoryRestoreFailure("invalid_snapshot_entry"); + } else if (entry.type !== "symlink" || typeof entry.target !== "string" || entry.target.length === 0 || entry.target.includes("\0")) { + throw new TaskDirectoryRestoreFailure("invalid_snapshot_entry"); + } + kinds.set(entry.path, entry.type); + } + if (manifest.excluded.some(value => !validRelativePath(value))) throw new TaskDirectoryRestoreFailure("invalid_snapshot_entry"); + const bodies = payload.subarray(start + length); + if (bodies.length !== bodyBytes) throw new TaskDirectoryRestoreFailure("invalid_snapshot_payload"); + return { manifest, bodies }; +} + +export function shortSiblingName(target: string, prefix: string) { + const length = Math.max(8, path.basename(target).length - prefix.length); + return path.join(path.dirname(target), `${prefix}${randomBytes(Math.ceil(length / 2)).toString("hex").slice(0, length)}`); +} + +export async function materializeTaskDirectorySnapshot(input: { payload: Buffer; target: string; expectedDigest: string; signal?: AbortSignal }) { + const { manifest, bodies } = decodeTaskDirectorySnapshot(input.payload); + if (taskDirectoryManifestDigest(manifest) !== input.expectedDigest) throw new TaskDirectoryRestoreFailure("snapshot_digest_mismatch"); + const staging = shortSiblingName(input.target, ".restoring-"); + await fs.mkdir(staging, { mode: 0o700 }); + try { + let offset = 0; + const directories: Array<{ target: string; mode: number }> = []; + for (const entry of manifest.entries) { + input.signal?.throwIfAborted(); + const target = path.join(staging, ...entry.path.split("/")); + if (entry.type === "directory") { + await fs.mkdir(target, { mode: 0o700 }); + directories.push({ target, mode: entry.mode }); + } else if (entry.type === "symlink") { + await fs.symlink(entry.target, target); + } else { + const body = bodies.subarray(offset, offset + entry.size); + offset += entry.size; + if (createHash("sha256").update(body).digest("hex") !== entry.sha256) throw new TaskDirectoryRestoreFailure("snapshot_content_mismatch"); + const handle = await fs.open(target, fs.constants.O_WRONLY | fs.constants.O_CREAT | fs.constants.O_EXCL | fs.constants.O_NOFOLLOW, 0o600); + try { await handle.writeFile(body); } finally { await handle.close(); } + await fs.chmod(target, entry.mode); + } + } + for (const directory of directories.reverse()) await fs.chmod(directory.target, directory.mode); + const restored = await inspectTaskDirectory(staging, { limits: { maxFiles: Number.MAX_SAFE_INTEGER, maxFileBytes: Number.MAX_SAFE_INTEGER, + maxTotalBytes: Number.MAX_SAFE_INTEGER }, signal: input.signal }); + if (restored.digest !== input.expectedDigest) throw new TaskDirectoryRestoreFailure("restored_contents_mismatch"); + return staging; + } catch (error) { + await removeTaskDirectoryTree(staging).catch(() => undefined); + throw error; + } +} + +export async function removeTaskDirectoryTree(target: string) { + const stat = await fs.lstat(target).catch((error: NodeJS.ErrnoException) => { + if (error.code === "ENOENT") return null; + throw error; + }); + if (!stat) return; + if (stat.isSymbolicLink() || !stat.isDirectory()) throw new Error("unsafe_task_directory_trash"); + const makeWritable = async (directory: string) => { + await fs.chmod(directory, 0o700); + for (const entry of await fs.readdir(directory, { withFileTypes: true })) { + if (entry.isDirectory()) await makeWritable(path.join(directory, entry.name)); + } + }; + await makeWritable(target).catch(() => undefined); + await fs.rm(target, { recursive: true, force: true }); +}