Sync native tracing with the retained controller base

Co-Authored-By: Paperclip <noreply@paperclip.ing>

* codex/runner-controller-evidence:
  Reclaim finished isolated task directories with restore points (#17472)
  Read execution blocker evidence without full execution payloads (#17491)
This commit is contained in:
DottaandPaperclip committed 2026-10-24 14:30:48 -07:00
commit 1e41d43c87
18 files changed
+2149 -75

No files matched your search

+6
View File
@@ -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
+11
View File
@@ -2624,6 +2624,17 @@ Environment overrides:
`PAPERCLIP_GIT_RESTORE_POINT_IDLE_DAYS=<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 `<instance>/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=<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=<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
+125
View File
@@ -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
`<instance>/isolated-workspaces/<companyId>/<issueId>/`. 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/<workspace>/<generation>/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-<random>` 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-<random>` 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
@@ -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<typeof createDb>;
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
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<typeof heartbeatService>, 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();
});
});
@@ -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 });
});
});
@@ -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<typeof createDb>;
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
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<typeof executionWorkspaceService>[1] = {}) => executionWorkspaceService(db, {
taskDirectoryRetention: settings, gitRestorePoints: restorePoints(), ...overrides });
const expire = async (overrides: Partial<Parameters<typeof expireIsolatedTaskDirectory>[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<ReturnType<typeof fixture>>) => Promise<unknown>;
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<Parameters<typeof sweepIsolatedTaskDirectories>[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);
});
});
});
+7
View File
@@ -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"
+5
View File
@@ -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
@@ -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<typeof heartbeatRuns.$inferSelect,
"id" | "companyId" | "status" | "startedAt" | "finishedAt" | "runtimeMode" |
"runtimeModeResolvedAt" | "errorCode" | "error" | "resultJson" | "runnerProfileJson" |
"contextSnapshot" | "processPid" | "processGroupId" | "processStartedAt" |
"sessionIdAfter" | "nativeIssueId" | "nativeSessionId" | "executionStage" |
"controllerBootId" | "controllerLeaseExpiresAt">;
type Coordinator = typeof nativeRunFinalizations.$inferSelect;
/** Candidate for explicit user Retry only, never automatic replay after Stop. */
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
let db: ReturnType<typeof createDb>;
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<typeof heartbeatRuns.$inferInsert> = {}, extraEvidence: Record<string, unknown> = {}) {
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<Db["select"]>) => {
const builder = target.select(...fields), from = builder.from.bind(builder);
builder.from = ((...args: Parameters<typeof from>) => {
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<typeof execute>) => {
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<ReturnType<typeof fixture>>) {
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<string, unknown> | null });
expect(await compare(f)).toMatchObject({ canRetry: true });
});
});
+47 -1
View File
@@ -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<Record<string, unknown> | 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<Record<string, unknown> | 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<boolean> {
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));
+164 -9
View File
@@ -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<boolean>`(${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<string, number>();
let taskDirectoryWalkDueAt = 0;
let taskDirectoryWalkCursor: string | null = null;
let taskDirectoryWalkRetryAt: number | null = null;
const taskDirectoryFolderDue = new Map<string, number>();
const taskDirectoryHints = new Set<string>();
const taskDirectoryRowSkips = new Map<string, { updatedAt: number; retryAt: number | null; issueId: string | null }>();
const TASK_DIRECTORY_FOLDERS_PER_SWEEP = 8;
const TASK_DIRECTORY_HINTS_PER_SWEEP = 64;
const remember = <K, V>(map: Map<K, V>, 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<ReturnType<typeof inspectIsolatedTaskDirectoryFolders>> = [];
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<boolean>`${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<string, ReturnType<typeof inspectGitCloseReadiness>>();
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<string>) => {
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<ReopenClosedIsolatedExecutionWorkspaceResult> => {
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<string, unknown>, 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" };
@@ -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 })
@@ -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<string> {
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<string, any>, objectKey: string, at: string) {
const expire = (ref: Record<string, unknown> | 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<string, any>) =>
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<string, any> | undefined;
if (!recipe) return;
const expire = (ref: Record<string, unknown> | 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<string, any>) =>
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<string, any> | 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");
}
@@ -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 (
@@ -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<typeof readGitCacheExpiration>[0])?.workspaceRemoved) {
&& (readGitCacheExpiration(existingExecutionWorkspace as Parameters<typeof readGitCacheExpiration>[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");
@@ -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<string, unknown> | 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<Workspace, "metadata" | "status">) {
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<Workspace, "sourceIssueId" | "strategyType" | "providerType" | "repoUrl" | "branchName" | "cwd" | "providerRef" | "mode">) {
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<Workspace, "companyId" | "sourceIssueId" | "strategyType" | "providerType" | "repoUrl" | "branchName" | "cwd" | "providerRef" | "mode">) {
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<Parameters<Db["transaction"]>[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<ReturnType<typeof captureTaskDirectorySnapshot>>, 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<void>;
}): Promise<TaskDirectoryRetentionResult> {
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<string, unknown> = { ...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<string, unknown> = { ...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<Parameters<Db["transaction"]>[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<Parameters<Db["transaction"]>[0]>[0], workspace: { id: string; metadata?: Record<string, unknown> | null }) {
const state = readTaskDirectoryRetention(workspace);
if (state) await abandonTaskDirectoryRestorePoint(tx, workspace.id, state.restorePoint.objectKey);
}
export async function restoreIsolatedTaskDirectory(input: {
storage: StorageService;
workspace: Pick<Workspace, "id" | "companyId" | "sourceIssueId" | "strategyType" | "providerType" | "repoUrl" | "branchName" | "cwd" | "providerRef" | "mode" | "metadata">;
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<IssueTree> {
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<void>;
};
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<TaskDirectoryFolderResult | null> {
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<string, { companyId: string; issueId: string }>();
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'`);
}
@@ -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<string>;
};
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<TaskDirectorySnapshotManifest, "entries">) {
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<number> {
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<string, TaskDirectoryEntry["type"]>();
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 });
}