diff --git a/server/src/__tests__/execution-workspace-reopen.test.ts b/server/src/__tests__/execution-workspace-reopen.test.ts new file mode 100644 index 0000000000..ea12f68e85 --- /dev/null +++ b/server/src/__tests__/execution-workspace-reopen.test.ts @@ -0,0 +1,462 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { + activityLog, + companies, + createDb, + executionWorkspaces, + issues, + projectWorkspaces, + projects, +} from "@paperclipai/db"; +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "./helpers/embedded-postgres.js"; +import { + EXECUTION_WORKSPACE_LIFECYCLE_GENERATION_METADATA_KEY, + EXECUTION_WORKSPACE_REOPEN_FAILED_REASON, + EXECUTION_WORKSPACE_REOPEN_PENDING_METADATA_KEY, + EXECUTION_WORKSPACE_REOPEN_PENDING_SINCE_METADATA_KEY, + executionWorkspaceService, + metadataHasReopenPendingConsumption, + readExecutionWorkspaceLifecycleGeneration, + readMetadataReopenPendingConsumptionSince, +} from "../services/execution-workspaces.js"; + +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; + +if (!embeddedPostgresSupport.supported) { + console.warn( + `Skipping embedded Postgres reopen tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`, + ); +} + +describeEmbeddedPostgres("reopen archived isolated execution workspace", () => { + let db!: ReturnType; + let tempDb: Awaited> | null = null; + const tempDirs: string[] = []; + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-reopen-"); + db = createDb(tempDb.connectionString); + }, 20_000); + + afterEach(async () => { + await db.delete(activityLog); + await db.delete(issues); + await db.delete(executionWorkspaces); + await db.delete(projectWorkspaces); + await db.delete(projects); + await db.delete(companies); + await Promise.all(tempDirs.splice(0).map((dir) => rm(dir, { recursive: true, force: true }))); + }); + + afterAll(async () => { + await tempDb?.cleanup(); + }); + + async function makeExistingDir(): Promise { + const dir = await mkdtemp(join(tmpdir(), "paperclip-reopen-cwd-")); + tempDirs.push(dir); + return dir; + } + + async function seedCompanyProject() { + const companyId = randomUUID(); + const projectId = randomUUID(); + const projectWorkspaceId = randomUUID(); + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix: `PAP-${companyId.slice(0, 8)}`, + requireBoardApprovalForNewAgents: false, + }); + await db.insert(projects).values({ + id: projectId, + companyId, + name: "Reopen project", + status: "in_progress", + }); + await db.insert(projectWorkspaces).values({ + id: projectWorkspaceId, + companyId, + projectId, + name: "Primary", + sourceType: "local_path", + cwd: "/tmp/paperclip-reopen-project", + isPrimary: true, + }); + return { companyId, projectId, projectWorkspaceId }; + } + + async function seedClosedWorkspace(input: { + companyId: string; + projectId: string; + projectWorkspaceId: string; + cwd: string; + status?: "archived" | "cleanup_failed" | "active"; + generation?: number; + }) { + const workspaceId = randomUUID(); + const closed = (input.status ?? "archived") !== "active"; + await db.insert(executionWorkspaces).values({ + id: workspaceId, + companyId: input.companyId, + projectId: input.projectId, + projectWorkspaceId: input.projectWorkspaceId, + mode: "isolated_workspace", + // project_primary strategy so the rebuild only checks that the directory + // exists, with no git operation. + strategyType: "project_primary", + name: "reopen-workspace", + status: input.status ?? "archived", + providerType: "local_fs", + cwd: input.cwd, + closedAt: closed ? new Date() : null, + cleanupReason: closed ? "issue_terminal" : null, + cleanupEligibleAt: closed ? new Date() : null, + metadata: { + [EXECUTION_WORKSPACE_LIFECYCLE_GENERATION_METADATA_KEY]: input.generation ?? 1, + }, + }); + return workspaceId; + } + + async function seedIssue(input: { + companyId: string; + projectId: string; + workspaceId: string; + issueNumber: number; + }) { + const issueId = randomUUID(); + await db.insert(issues).values({ + id: issueId, + companyId: input.companyId, + projectId: input.projectId, + identifier: `PAP-${input.issueNumber}`, + issueNumber: input.issueNumber, + title: "Resume me", + status: "todo", + priority: "medium", + executionWorkspaceId: input.workspaceId, + }); + return issueId; + } + + async function readWorkspace(id: string) { + return db + .select() + .from(executionWorkspaces) + .where(eq(executionWorkspaces.id, id)) + .then((rows) => rows[0] ?? null); + } + + it("reopens the archived row in place, keeps the issue link, and raises the generation", async () => { + const { companyId, projectId, projectWorkspaceId } = await seedCompanyProject(); + const cwd = await makeExistingDir(); + const workspaceId = await seedClosedWorkspace({ companyId, projectId, projectWorkspaceId, cwd, generation: 3 }); + const issueId = await seedIssue({ companyId, projectId, workspaceId, issueNumber: 4100 }); + + const svc = executionWorkspaceService(db); + const result = await svc.reopenClosedIsolatedExecutionWorkspaceForIssue({ + workspaceId, + issue: { id: issueId, companyId, projectId }, + actor: { agentId: null, actorType: "user" }, + }); + + expect(result.ok).toBe(true); + if (!result.ok) return; + expect(result.reopened).toBe(true); + + const row = await readWorkspace(workspaceId); + expect(row?.status).toBe("active"); + expect(row?.closedAt).toBeNull(); + expect(row?.cleanupReason).toBeNull(); + expect(row?.cleanupEligibleAt).toBeNull(); + expect(readExecutionWorkspaceLifecycleGeneration(row?.metadata as Record | null)).toBe(4); + // The reopen flags the row so the terminal reaper does not archive and + // destroy the rebuilt worktree before the caller consumes it. + expect(metadataHasReopenPendingConsumption(row?.metadata as Record | null)).toBe(true); + + // The reopen never changes the issue-to-workspace link. + const issueRow = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null); + expect(issueRow?.executionWorkspaceId).toBe(workspaceId); + }); + + it("keeps access for every issue that shares the reopened row", async () => { + const { companyId, projectId, projectWorkspaceId } = await seedCompanyProject(); + const cwd = await makeExistingDir(); + const workspaceId = await seedClosedWorkspace({ companyId, projectId, projectWorkspaceId, cwd }); + const firstIssueId = await seedIssue({ companyId, projectId, workspaceId, issueNumber: 4101 }); + const secondIssueId = await seedIssue({ companyId, projectId, workspaceId, issueNumber: 4102 }); + + const svc = executionWorkspaceService(db); + const result = await svc.reopenClosedIsolatedExecutionWorkspaceForIssue({ + workspaceId, + issue: { id: firstIssueId, companyId, projectId }, + actor: { agentId: null, actorType: "user" }, + }); + expect(result.ok).toBe(true); + + const rows = await db.select().from(issues); + for (const row of rows) { + expect(row.executionWorkspaceId).toBe(workspaceId); + } + const workspace = await readWorkspace(workspaceId); + expect(workspace?.status).toBe("active"); + expect(firstIssueId).not.toBe(secondIssueId); + }); + + it("fails closed and keeps the row closed when the rebuild fails", async () => { + const { companyId, projectId, projectWorkspaceId } = await seedCompanyProject(); + // A directory that does not exist. The project_primary rebuild returns null. + const missingDir = join(tmpdir(), `paperclip-reopen-missing-${randomUUID()}`); + const workspaceId = await seedClosedWorkspace({ + companyId, + projectId, + projectWorkspaceId, + cwd: missingDir, + generation: 2, + }); + const issueId = await seedIssue({ companyId, projectId, workspaceId, issueNumber: 4103 }); + + const svc = executionWorkspaceService(db); + const result = await svc.reopenClosedIsolatedExecutionWorkspaceForIssue({ + workspaceId, + issue: { id: issueId, companyId, projectId }, + actor: { agentId: null, actorType: "user" }, + }); + + expect(result.ok).toBe(false); + if (result.ok) return; + expect(result.code).toBe("rebuild_failed"); + + const row = await readWorkspace(workspaceId); + // The row stays closed and retryable. The generation still rises so a queued + // cleanup with the old generation does nothing. + expect(row?.status).toBe("archived"); + expect(row?.cleanupReason).toBe(EXECUTION_WORKSPACE_REOPEN_FAILED_REASON); + expect(row?.cleanupEligibleAt).toBeNull(); + expect(readExecutionWorkspaceLifecycleGeneration(row?.metadata as Record | null)).toBe(3); + }); + + it("refuses to reopen a workspace in another company", async () => { + const first = await seedCompanyProject(); + const second = await seedCompanyProject(); + const cwd = await makeExistingDir(); + const workspaceId = await seedClosedWorkspace({ + companyId: first.companyId, + projectId: first.projectId, + projectWorkspaceId: first.projectWorkspaceId, + cwd, + }); + + const svc = executionWorkspaceService(db); + const result = await svc.reopenClosedIsolatedExecutionWorkspaceForIssue({ + workspaceId, + // The issue belongs to a different company than the workspace. + issue: { id: randomUUID(), companyId: second.companyId, projectId: second.projectId }, + actor: { agentId: null, actorType: "user" }, + }); + + expect(result.ok).toBe(false); + if (result.ok) return; + expect(result.code).toBe("not_reopenable"); + + const row = await readWorkspace(workspaceId); + expect(row?.status).toBe("archived"); + }); + + it("reports success without a rebuild when the workspace is already active", async () => { + const { companyId, projectId, projectWorkspaceId } = await seedCompanyProject(); + const cwd = await makeExistingDir(); + const workspaceId = await seedClosedWorkspace({ + companyId, + projectId, + projectWorkspaceId, + cwd, + status: "active", + generation: 5, + }); + const issueId = await seedIssue({ companyId, projectId, workspaceId, issueNumber: 4104 }); + + const svc = executionWorkspaceService(db); + const result = await svc.reopenClosedIsolatedExecutionWorkspaceForIssue({ + workspaceId, + issue: { id: issueId, companyId, projectId }, + actor: { agentId: null, actorType: "user" }, + }); + + expect(result.ok).toBe(true); + if (!result.ok) return; + expect(result.reopened).toBe(false); + + const row = await readWorkspace(workspaceId); + // No reopen ran, so the generation is unchanged. + expect(readExecutionWorkspaceLifecycleGeneration(row?.metadata as Record | null)).toBe(5); + }); + + it("runs the destroy under the fence when the generation matches, and skips it after a reopen", async () => { + const { companyId, projectId, projectWorkspaceId } = await seedCompanyProject(); + const cwd = await makeExistingDir(); + const workspaceId = await seedClosedWorkspace({ companyId, projectId, projectWorkspaceId, cwd, generation: 7 }); + + const svc = executionWorkspaceService(db); + + // The captured generation matches, so the destroy callback runs. + const destroyMatches = vi.fn(async () => "destroyed"); + const matched = await svc.fenceClosedWorkspaceDestruction({ + workspaceId, + capturedGeneration: 7, + destroy: destroyMatches, + }); + expect(matched.skippedReopened).toBe(false); + expect(destroyMatches).toHaveBeenCalledTimes(1); + + // Simulate a reopen: raise the generation and mark the row active. + await db + .update(executionWorkspaces) + .set({ + status: "active", + closedAt: null, + cleanupReason: null, + cleanupEligibleAt: null, + metadata: { [EXECUTION_WORKSPACE_LIFECYCLE_GENERATION_METADATA_KEY]: 8 }, + }) + .where(eq(executionWorkspaces.id, workspaceId)); + + // A stale cleanup captured generation 7. The fence must skip the destroy. + const destroyStale = vi.fn(async () => "destroyed"); + const skipped = await svc.fenceClosedWorkspaceDestruction({ + workspaceId, + capturedGeneration: 7, + destroy: destroyStale, + }); + expect(skipped.skippedReopened).toBe(true); + expect(destroyStale).not.toHaveBeenCalled(); + }); + + it("clears the reopen-pending flag after an unconsumed reopen and stays idempotent", async () => { + const { companyId, projectId, projectWorkspaceId } = await seedCompanyProject(); + const cwd = await makeExistingDir(); + const workspaceId = await seedClosedWorkspace({ companyId, projectId, projectWorkspaceId, cwd }); + const issueId = await seedIssue({ companyId, projectId, workspaceId, issueNumber: 4109 }); + + const svc = executionWorkspaceService(db); + // The reopen publishes the row as active and sets the reopen-pending flag. + const reopenResult = await svc.reopenClosedIsolatedExecutionWorkspaceForIssue({ + workspaceId, + issue: { id: issueId, companyId, projectId }, + actor: { agentId: null, actorType: "user" }, + }); + expect(reopenResult.ok).toBe(true); + if (!reopenResult.ok) throw new Error("reopen failed"); + const reopenGeneration = reopenResult.generation; + const reopenedRow = await readWorkspace(workspaceId); + expect(metadataHasReopenPendingConsumption(reopenedRow?.metadata as Record | null)).toBe(true); + // The reopen stamps the time it set the flag, so the reaper can age a + // stranded flag out of the way after the grace period. + expect(readMetadataReopenPendingConsumptionSince(reopenedRow?.metadata as Record | null)) + .toBeInstanceOf(Date); + + // The caller never consumed the reopen, so clear the flag at its generation. + const cleared = await svc.clearReopenPendingConsumptionForUnconsumedReopen({ + workspaceId, + issue: { id: issueId, companyId }, + actor: { agentId: null, actorType: "user" }, + expectedGeneration: reopenGeneration, + }); + expect(cleared.cleared).toBe(true); + + const clearedRow = await readWorkspace(workspaceId); + // The flag is gone, so the terminal reaper can archive and reclaim the row. + expect(metadataHasReopenPendingConsumption(clearedRow?.metadata as Record | null)).toBe(false); + // The clear removes the timestamp too, so no orphan key survives. + expect((clearedRow?.metadata as Record | null)?.[EXECUTION_WORKSPACE_REOPEN_PENDING_SINCE_METADATA_KEY]) + .toBeUndefined(); + // The row stays active, so a retried resume can still reuse the rebuilt worktree. + expect(clearedRow?.status).toBe("active"); + + const events = await db + .select() + .from(activityLog) + .where(eq(activityLog.entityId, workspaceId)); + expect(events.some((event) => event.action === "execution_workspace.reopen_unconsumed")).toBe(true); + + // A second call finds no flag and does nothing. + const second = await svc.clearReopenPendingConsumptionForUnconsumedReopen({ + workspaceId, + issue: { id: issueId, companyId }, + actor: { agentId: null, actorType: "user" }, + expectedGeneration: reopenGeneration, + }); + expect(second.cleared).toBe(false); + const eventsAfter = await db + .select() + .from(activityLog) + .where(eq(activityLog.entityId, workspaceId)); + expect(eventsAfter.filter((event) => event.action === "execution_workspace.reopen_unconsumed").length).toBe(1); + }); + + it("does not clear a newer reopen's fence when a stale request presents an old generation", async () => { + const { companyId, projectId, projectWorkspaceId } = await seedCompanyProject(); + const cwd = await makeExistingDir(); + const workspaceId = await seedClosedWorkspace({ companyId, projectId, projectWorkspaceId, cwd }); + const issueId = await seedIssue({ companyId, projectId, workspaceId, issueNumber: 4115 }); + + const svc = executionWorkspaceService(db); + const reopenResult = await svc.reopenClosedIsolatedExecutionWorkspaceForIssue({ + workspaceId, + issue: { id: issueId, companyId, projectId }, + actor: { agentId: null, actorType: "user" }, + }); + expect(reopenResult.ok).toBe(true); + if (!reopenResult.ok) throw new Error("reopen failed"); + const staleGeneration = reopenResult.generation; + + // Simulate a newer reopen that raised the generation and installed its own + // fence with a fresh timestamp. This models two overlapping reopen requests. + const newerGeneration = staleGeneration + 1; + await db + .update(executionWorkspaces) + .set({ + metadata: { + [EXECUTION_WORKSPACE_LIFECYCLE_GENERATION_METADATA_KEY]: newerGeneration, + [EXECUTION_WORKSPACE_REOPEN_PENDING_METADATA_KEY]: true, + [EXECUTION_WORKSPACE_REOPEN_PENDING_SINCE_METADATA_KEY]: new Date().toISOString(), + }, + }) + .where(eq(executionWorkspaces.id, workspaceId)); + + // The stale request's response-end clear presents the old generation. It must + // not clear the newer reopen's live fence. + const staleClear = await svc.clearReopenPendingConsumptionForUnconsumedReopen({ + workspaceId, + issue: { id: issueId, companyId }, + actor: { agentId: null, actorType: "user" }, + expectedGeneration: staleGeneration, + }); + expect(staleClear.cleared).toBe(false); + const afterStale = await readWorkspace(workspaceId); + expect(metadataHasReopenPendingConsumption(afterStale?.metadata as Record | null)).toBe(true); + expect(readExecutionWorkspaceLifecycleGeneration(afterStale?.metadata as Record | null)) + .toBe(newerGeneration); + + // The newer owner clears its own fence at the matching generation. + const ownerClear = await svc.clearReopenPendingConsumptionForUnconsumedReopen({ + workspaceId, + issue: { id: issueId, companyId }, + actor: { agentId: null, actorType: "user" }, + expectedGeneration: newerGeneration, + }); + expect(ownerClear.cleared).toBe(true); + const afterOwner = await readWorkspace(workspaceId); + expect(metadataHasReopenPendingConsumption(afterOwner?.metadata as Record | null)).toBe(false); + }); +}); diff --git a/server/src/__tests__/execution-workspaces-routes.test.ts b/server/src/__tests__/execution-workspaces-routes.test.ts index be7deacb2e..d7f9913ce5 100644 --- a/server/src/__tests__/execution-workspaces-routes.test.ts +++ b/server/src/__tests__/execution-workspaces-routes.test.ts @@ -10,6 +10,8 @@ const mockExecutionWorkspaceService = vi.hoisted(() => ({ listSummaries: vi.fn(), getById: vi.fn(), getCloseReadiness: vi.fn(), + archiveWorkspaceUnderLifecycleLock: vi.fn(), + fenceClosedWorkspaceDestruction: vi.fn(), reconcileExecutionWorkspaceBranch: vi.fn(), update: vi.fn(), })); @@ -28,6 +30,10 @@ const mockAccessService = vi.hoisted(() => ({ })); const mockLogActivity = vi.hoisted(() => vi.fn(async () => undefined)); +const mockEnvironmentRuntimeService = vi.hoisted(() => ({ + destroyReusableSandboxLeases: vi.fn(async () => undefined), +})); + vi.mock("../services/index.js", () => ({ accessService: () => mockAccessService, executionWorkspaceService: () => mockExecutionWorkspaceService, @@ -36,6 +42,25 @@ vi.mock("../services/index.js", () => ({ workspaceOperationService: () => mockWorkspaceOperationService, })); +vi.mock("../services/environment-runtime.js", () => ({ + environmentRuntimeService: () => mockEnvironmentRuntimeService, +})); + +const mockWorkspaceRuntimeTeardown = vi.hoisted(() => ({ + stopRuntimeServicesForExecutionWorkspace: vi.fn(async () => undefined), + cleanupExecutionWorkspaceArtifacts: vi.fn(async () => ({ cleaned: true, warnings: [] as string[] })), +})); + +vi.mock("../services/workspace-runtime.js", async (importActual) => { + const actual = await importActual(); + return { + ...actual, + stopRuntimeServicesForExecutionWorkspace: + mockWorkspaceRuntimeTeardown.stopRuntimeServicesForExecutionWorkspace, + cleanupExecutionWorkspaceArtifacts: mockWorkspaceRuntimeTeardown.cleanupExecutionWorkspaceArtifacts, + }; +}); + function createApp(actor: Record = { type: "board", userId: "local-board", @@ -389,4 +414,126 @@ describe.sequential("execution workspace routes", () => { }), })); }); + + it("returns 409 and skips destructive cleanup when the archive hits a reopen-pending workspace", async () => { + // A reopen published the workspace active while its source issue is still + // terminal. The archive control must return 409 before any lease teardown, + // runtime-service stop, or artifact cleanup, so it never removes the rebuilt + // worktree. + mockExecutionWorkspaceService.getById.mockResolvedValue({ + id: "workspace-1", + companyId: "company-1", + sourceIssueId: "issue-1", + status: "active", + mode: "isolated_workspace", + }); + mockExecutionWorkspaceService.getCloseReadiness.mockResolvedValue({ + state: "ready", + blockingReasons: [], + }); + mockExecutionWorkspaceService.archiveWorkspaceUnderLifecycleLock.mockResolvedValue({ + outcome: "reopen_pending", + }); + + const res = await request(createApp()) + .patch("/api/execution-workspaces/workspace-1") + .send({ status: "archived" }); + + expect(res.status).toBe(409); + expect(mockExecutionWorkspaceService.archiveWorkspaceUnderLifecycleLock).toHaveBeenCalledTimes(1); + // The destruction fence never runs, so no worktree is removed. + expect(mockExecutionWorkspaceService.fenceClosedWorkspaceDestruction).not.toHaveBeenCalled(); + }); + + it("destroys the reusable sandbox leases inside the destruction fence when the archive wins", async () => { + // The archive wins the lifecycle race. The fence runs the destroy callback, + // so the reusable sandbox lease teardown runs with the worktree teardown. + const archivedWorkspace = { + id: "workspace-1", + companyId: "company-1", + sourceIssueId: "issue-1", + status: "archived", + mode: "isolated_workspace", + projectWorkspaceId: null, + projectId: null, + cwd: "/tmp/worktree", + }; + mockExecutionWorkspaceService.getById.mockResolvedValue({ + ...archivedWorkspace, + status: "active", + }); + mockExecutionWorkspaceService.getCloseReadiness.mockResolvedValue({ + state: "ready", + blockingReasons: [], + }); + mockExecutionWorkspaceService.archiveWorkspaceUnderLifecycleLock.mockResolvedValue({ + outcome: "archived", + workspace: archivedWorkspace, + capturedGeneration: 3, + }); + mockExecutionWorkspaceService.fenceClosedWorkspaceDestruction.mockImplementation( + async ({ destroy }: { destroy: () => Promise }) => ({ + skippedReopened: false, + result: await destroy(), + }), + ); + + const res = await request(createApp()) + .patch("/api/execution-workspaces/workspace-1") + .send({ status: "archived" }); + + expect(res.status).toBe(200); + // The lease teardown runs inside the fence, so it uses the closed-workspace + // failure reason and targets the archived row. + expect(mockEnvironmentRuntimeService.destroyReusableSandboxLeases).toHaveBeenCalledWith( + expect.objectContaining({ + companyId: "company-1", + executionWorkspaceId: "workspace-1", + failureReason: "execution_workspace_closed", + }), + ); + }); + + it("keeps the reusable sandbox leases when a reopen makes the fence skip the archive teardown", async () => { + // A reopen raised the lifecycle generation after the archive captured its + // own generation. The fence skips the destroy callback and keeps the + // reopened row, so the lease teardown must not run. Before the fix the lease + // teardown ran before the fence, so an overlapping reopen lost its leases. + const archivedWorkspace = { + id: "workspace-1", + companyId: "company-1", + sourceIssueId: "issue-1", + status: "archived", + mode: "isolated_workspace", + projectWorkspaceId: null, + projectId: null, + cwd: "/tmp/worktree", + }; + mockExecutionWorkspaceService.getById.mockResolvedValue({ + ...archivedWorkspace, + status: "active", + }); + mockExecutionWorkspaceService.getCloseReadiness.mockResolvedValue({ + state: "ready", + blockingReasons: [], + }); + mockExecutionWorkspaceService.archiveWorkspaceUnderLifecycleLock.mockResolvedValue({ + outcome: "archived", + workspace: archivedWorkspace, + capturedGeneration: 3, + }); + // The fence detects the reopen and never runs the destroy callback. + mockExecutionWorkspaceService.fenceClosedWorkspaceDestruction.mockResolvedValue({ + skippedReopened: true, + }); + + const res = await request(createApp()) + .patch("/api/execution-workspaces/workspace-1") + .send({ status: "archived" }); + + expect(res.status).toBe(200); + expect(mockExecutionWorkspaceService.fenceClosedWorkspaceDestruction).toHaveBeenCalledTimes(1); + // The reopen keeps its reusable leases because the fence skipped the destroy. + expect(mockEnvironmentRuntimeService.destroyReusableSandboxLeases).not.toHaveBeenCalled(); + }); }); diff --git a/server/src/__tests__/execution-workspaces-service.test.ts b/server/src/__tests__/execution-workspaces-service.test.ts index 1bb8db49f3..5c5a239897 100644 --- a/server/src/__tests__/execution-workspaces-service.test.ts +++ b/server/src/__tests__/execution-workspaces-service.test.ts @@ -28,10 +28,15 @@ import { startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; import { + EXECUTION_WORKSPACE_LIFECYCLE_GENERATION_METADATA_KEY, + EXECUTION_WORKSPACE_REOPEN_PENDING_METADATA_KEY, + EXECUTION_WORKSPACE_REOPEN_PENDING_SINCE_METADATA_KEY, executionWorkspaceService, deriveExecutionWorkspaceDeliveryState, mergeExecutionWorkspaceConfig, + metadataHasReopenPendingConsumption, readExecutionWorkspaceConfig, + readMetadataReopenPendingConsumptionSince, } from "../services/execution-workspaces.ts"; import { issueService } from "../services/issues.ts"; import { @@ -929,6 +934,54 @@ describeEmbeddedPostgres("executionWorkspaceService.getCloseReadiness", () => { .resolves.toBe("not delivered\n"); }); + it("does not write stale cleanup-failure state onto a newer archive lifecycle", async () => { + // Reproduce the cleanup-failure race. The reaper archives the workspace at one + // generation and captures it. The cleanup then throws. Before the catch handler + // writes the cleanup-failed status, a reopen and a fresh archive raise the + // generation. The catch handler must skip its write, so the stale failure never + // overwrites the newer archive lifecycle. + const seeded = await seedTerminalWorkspace({ mergedPr: true }); + const newerReason = "newer_archive_lifecycle_marker"; + const racingService = executionWorkspaceService(db, { + resolvePullRequestDetails: async (_companyId, reference) => + pullRequestDetailsByKey.get(`${seeded.companyId}:${reference.number}`) ?? { state: "unknown" }, + beforeTerminalWorkspaceCleanup: async (workspace) => { + // Stand in for a reopen and a fresh archive that ran after this sweep + // captured the generation. Raise the generation past the captured value, + // keep the row closed, then force the cleanup to throw. + await db + .update(executionWorkspaces) + .set({ + status: "archived", + cleanupReason: newerReason, + metadata: { [EXECUTION_WORKSPACE_LIFECYCLE_GENERATION_METADATA_KEY]: 2 }, + updatedAt: new Date(), + }) + .where(eq(executionWorkspaces.id, workspace.id)); + throw new Error("forced cleanup failure"); + }, + }); + + const sweep = await racingService.sweepTerminalWorkspaces(); + const [workspace] = await db + .select({ + status: executionWorkspaces.status, + cleanupReason: executionWorkspaces.cleanupReason, + metadata: executionWorkspaces.metadata, + }) + .from(executionWorkspaces) + .where(eq(executionWorkspaces.id, seeded.executionWorkspaceId)); + + expect(sweep).toMatchObject({ cleanupFailed: 1 }); + // The fenced write saw the raised generation and skipped, so the newer + // lifecycle state survives untouched. + expect(workspace?.status).toBe("archived"); + expect(workspace?.cleanupReason).toBe(newerReason); + expect( + (workspace?.metadata as Record | null)?.[EXECUTION_WORKSPACE_LIFECYCLE_GENERATION_METADATA_KEY], + ).toBe(2); + }); + it("archives terminal workspaces without running configured cleanup hooks", async () => { const seeded = await seedTerminalWorkspace({ mergedPr: true }); const cleanupMarker = path.join(path.dirname(seeded.worktreePath), `cleanup-marker-${randomUUID()}`); @@ -952,6 +1005,581 @@ describeEmbeddedPostgres("executionWorkspaceService.getCloseReadiness", () => { await expect(fs.access(cleanupMarker)).rejects.toThrow(); }); + it("does not reap a reopened workspace while the source issue is still terminal", async () => { + // Reproduce the reverse-ordering race. A resume reopens the archived + // workspace and publishes it active, but the route has not yet changed the + // source issue out of the terminal state. The sweep must not archive and + // destroy the rebuilt worktree in this window. + const seeded = await seedTerminalWorkspace({ mergedPr: true }); + await db + .update(executionWorkspaces) + .set({ + status: "active", + closedAt: null, + cleanupReason: null, + cleanupEligibleAt: null, + metadata: { + createdByRuntime: true, + [EXECUTION_WORKSPACE_LIFECYCLE_GENERATION_METADATA_KEY]: 4, + [EXECUTION_WORKSPACE_REOPEN_PENDING_METADATA_KEY]: true, + // A fresh timestamp marks the reopen as in flight, so the sweep skips it. + [EXECUTION_WORKSPACE_REOPEN_PENDING_SINCE_METADATA_KEY]: new Date().toISOString(), + }, + }) + .where(eq(executionWorkspaces.id, seeded.executionWorkspaceId)); + + const sweep = await svc.sweepTerminalWorkspaces(); + const [workspace] = await db + .select({ status: executionWorkspaces.status, metadata: executionWorkspaces.metadata }) + .from(executionWorkspaces) + .where(eq(executionWorkspaces.id, seeded.executionWorkspaceId)); + + expect(sweep).toMatchObject({ archived: 0, skippedReopened: 1 }); + expect(workspace?.status).toBe("active"); + // The reopen flag stays until the source issue leaves the terminal state. + expect(metadataHasReopenPendingConsumption(workspace?.metadata as Record | null)).toBe(true); + // The rebuilt worktree is intact. + await expect(fs.access(seeded.worktreePath)).resolves.toBeUndefined(); + }); + + it("clears a stranded reopen flag whose consumer never ran, then reaps on a later sweep", async () => { + // A reopen published the workspace active and set the flag, but the consuming + // request never moved the source issue out of the terminal state, and the + // response-end clear never landed. The flag is older than the grace period. + // The first sweep clears the stranded flag; a later sweep archives the row. + const seeded = await seedTerminalWorkspace({ mergedPr: true }); + const strandedSince = new Date(Date.now() - 10 * 60 * 1000).toISOString(); + await db + .update(executionWorkspaces) + .set({ + status: "active", + closedAt: null, + cleanupReason: null, + cleanupEligibleAt: null, + metadata: { + createdByRuntime: true, + [EXECUTION_WORKSPACE_LIFECYCLE_GENERATION_METADATA_KEY]: 4, + [EXECUTION_WORKSPACE_REOPEN_PENDING_METADATA_KEY]: true, + [EXECUTION_WORKSPACE_REOPEN_PENDING_SINCE_METADATA_KEY]: strandedSince, + }, + }) + .where(eq(executionWorkspaces.id, seeded.executionWorkspaceId)); + + const firstSweep = await svc.sweepTerminalWorkspaces(); + const [afterClear] = await db + .select({ status: executionWorkspaces.status, metadata: executionWorkspaces.metadata }) + .from(executionWorkspaces) + .where(eq(executionWorkspaces.id, seeded.executionWorkspaceId)); + + // The first sweep clears the stranded flag but keeps the row active, so a + // retried resume can still reuse the rebuilt worktree. + expect(firstSweep).toMatchObject({ archived: 0, clearedStaleReopenPending: 1 }); + expect(afterClear?.status).toBe("active"); + expect(metadataHasReopenPendingConsumption(afterClear?.metadata as Record | null)).toBe(false); + await expect(fs.access(seeded.worktreePath)).resolves.toBeUndefined(); + + // A later sweep archives the reclaimed workspace through the normal path. + const secondSweep = await svc.sweepTerminalWorkspaces(); + const [afterArchive] = await db + .select({ status: executionWorkspaces.status }) + .from(executionWorkspaces) + .where(eq(executionWorkspaces.id, seeded.executionWorkspaceId)); + + expect(secondSweep).toMatchObject({ archived: 1 }); + expect(afterArchive?.status).toBe("archived"); + }); + + it("keeps the reopen fence for a request that outruns the grace period", async () => { + // A reopen published the workspace active and set the flag. The consuming + // request still runs, but it outran the grace period, so the flag looks + // stale by age. A live run owns the fence, so the sweep must not clear it. + // If the sweep cleared it, a later sweep could archive and destroy the + // rebuilt worktree under the running request. + const seeded = await seedTerminalWorkspace({ mergedPr: true, activeRun: true }); + const staleSince = new Date(Date.now() - 10 * 60 * 1000).toISOString(); + await db + .update(executionWorkspaces) + .set({ + status: "active", + closedAt: null, + cleanupReason: null, + cleanupEligibleAt: null, + metadata: { + createdByRuntime: true, + [EXECUTION_WORKSPACE_LIFECYCLE_GENERATION_METADATA_KEY]: 4, + [EXECUTION_WORKSPACE_REOPEN_PENDING_METADATA_KEY]: true, + [EXECUTION_WORKSPACE_REOPEN_PENDING_SINCE_METADATA_KEY]: staleSince, + }, + }) + .where(eq(executionWorkspaces.id, seeded.executionWorkspaceId)); + + const sweep = await svc.sweepTerminalWorkspaces(); + const [workspace] = await db + .select({ status: executionWorkspaces.status, metadata: executionWorkspaces.metadata }) + .from(executionWorkspaces) + .where(eq(executionWorkspaces.id, seeded.executionWorkspaceId)); + + // The live run holds the fence, so the sweep skips the workspace and keeps + // the flag. It clears nothing. + expect(sweep).toMatchObject({ archived: 0, skippedReopened: 1, clearedStaleReopenPending: 0 }); + expect(workspace?.status).toBe("active"); + expect(metadataHasReopenPendingConsumption(workspace?.metadata as Record | null)).toBe(true); + // The rebuilt worktree is intact. + await expect(fs.access(seeded.worktreePath)).resolves.toBeUndefined(); + }); + + it("refreshes the reopen fence for an in-flight request, so a later sweep keeps it", async () => { + // The consuming request is an HTTP request, not a heartbeat run, so the sweep + // cannot see it through the active-run check. The request re-stamps the flag on + // an interval below the grace period. This test drives one re-stamp on a flag + // that already looks stale by age. After the re-stamp the flag looks fresh, so + // the sweep skips the workspace and clears nothing, even with no active run. + const seeded = await seedTerminalWorkspace({ mergedPr: true }); + const staleSince = new Date(Date.now() - 10 * 60 * 1000).toISOString(); + await db + .update(executionWorkspaces) + .set({ + status: "active", + closedAt: null, + cleanupReason: null, + cleanupEligibleAt: null, + metadata: { + createdByRuntime: true, + [EXECUTION_WORKSPACE_LIFECYCLE_GENERATION_METADATA_KEY]: 4, + [EXECUTION_WORKSPACE_REOPEN_PENDING_METADATA_KEY]: true, + [EXECUTION_WORKSPACE_REOPEN_PENDING_SINCE_METADATA_KEY]: staleSince, + }, + }) + .where(eq(executionWorkspaces.id, seeded.executionWorkspaceId)); + + const result = await svc.refreshReopenPendingConsumption({ + workspaceId: seeded.executionWorkspaceId, + expectedGeneration: 4, + }); + expect(result).toEqual({ refreshed: true }); + + const [afterRefresh] = await db + .select({ metadata: executionWorkspaces.metadata }) + .from(executionWorkspaces) + .where(eq(executionWorkspaces.id, seeded.executionWorkspaceId)); + const refreshedSince = readMetadataReopenPendingConsumptionSince( + afterRefresh?.metadata as Record | null, + ); + // The re-stamp moved the timestamp forward, so the flag no longer looks stale. + expect(refreshedSince).not.toBeNull(); + expect(refreshedSince!.getTime()).toBeGreaterThan(new Date(staleSince).getTime()); + + const sweep = await svc.sweepTerminalWorkspaces(); + const [workspace] = await db + .select({ status: executionWorkspaces.status, metadata: executionWorkspaces.metadata }) + .from(executionWorkspaces) + .where(eq(executionWorkspaces.id, seeded.executionWorkspaceId)); + + // The fresh flag keeps the fence, so the sweep skips the workspace and clears + // nothing, even though no heartbeat run owns it. + expect(sweep).toMatchObject({ archived: 0, skippedReopened: 1, clearedStaleReopenPending: 0 }); + expect(workspace?.status).toBe("active"); + expect(metadataHasReopenPendingConsumption(workspace?.metadata as Record | null)).toBe(true); + await expect(fs.access(seeded.worktreePath)).resolves.toBeUndefined(); + }); + + it("does not refresh the reopen fence when a newer generation owns it", async () => { + // A newer reopen or an archive raised the generation, so the flag belongs to a + // new owner. A stale caller must not re-stamp another owner's fence. The + // refresh reports refreshed=false and leaves the timestamp unchanged. + const seeded = await seedTerminalWorkspace({ mergedPr: true }); + const since = new Date(Date.now() - 60 * 1000).toISOString(); + await db + .update(executionWorkspaces) + .set({ + status: "active", + closedAt: null, + cleanupReason: null, + cleanupEligibleAt: null, + metadata: { + createdByRuntime: true, + [EXECUTION_WORKSPACE_LIFECYCLE_GENERATION_METADATA_KEY]: 7, + [EXECUTION_WORKSPACE_REOPEN_PENDING_METADATA_KEY]: true, + [EXECUTION_WORKSPACE_REOPEN_PENDING_SINCE_METADATA_KEY]: since, + }, + }) + .where(eq(executionWorkspaces.id, seeded.executionWorkspaceId)); + + const result = await svc.refreshReopenPendingConsumption({ + workspaceId: seeded.executionWorkspaceId, + expectedGeneration: 4, + }); + expect(result).toEqual({ refreshed: false }); + + const [afterRefresh] = await db + .select({ metadata: executionWorkspaces.metadata }) + .from(executionWorkspaces) + .where(eq(executionWorkspaces.id, seeded.executionWorkspaceId)); + const unchangedSince = readMetadataReopenPendingConsumptionSince( + afterRefresh?.metadata as Record | null, + ); + expect(unchangedSince?.toISOString()).toBe(since); + }); + + it("does not refresh the reopen fence when the flag is already clear", async () => { + // The response-end clear already removed the flag. A late keepalive tick must + // not revive it. The refresh reports refreshed=false and adds no flag. + const seeded = await seedTerminalWorkspace({ mergedPr: true }); + await db + .update(executionWorkspaces) + .set({ + status: "active", + closedAt: null, + cleanupReason: null, + cleanupEligibleAt: null, + metadata: { + createdByRuntime: true, + [EXECUTION_WORKSPACE_LIFECYCLE_GENERATION_METADATA_KEY]: 4, + }, + }) + .where(eq(executionWorkspaces.id, seeded.executionWorkspaceId)); + + const result = await svc.refreshReopenPendingConsumption({ + workspaceId: seeded.executionWorkspaceId, + expectedGeneration: 4, + }); + expect(result).toEqual({ refreshed: false }); + + const [afterRefresh] = await db + .select({ metadata: executionWorkspaces.metadata }) + .from(executionWorkspaces) + .where(eq(executionWorkspaces.id, seeded.executionWorkspaceId)); + expect(metadataHasReopenPendingConsumption(afterRefresh?.metadata as Record | null)).toBe(false); + }); + + it("clears the reopen flag once the source issue leaves the terminal state", async () => { + // The resume transition committed, so the source issue is non-terminal. The + // sweep clears the stale reopen flag so a later terminal cycle can reap the + // workspace normally. + const seeded = await seedTerminalWorkspace({ mergedPr: true }); + await db + .update(executionWorkspaces) + .set({ + metadata: { + createdByRuntime: true, + [EXECUTION_WORKSPACE_LIFECYCLE_GENERATION_METADATA_KEY]: 4, + [EXECUTION_WORKSPACE_REOPEN_PENDING_METADATA_KEY]: true, + }, + }) + .where(eq(executionWorkspaces.id, seeded.executionWorkspaceId)); + await db + .update(issues) + .set({ status: "in_progress" }) + .where(eq(issues.id, seeded.sourceIssueId)); + + const sweep = await svc.sweepTerminalWorkspaces(); + const [workspace] = await db + .select({ status: executionWorkspaces.status, metadata: executionWorkspaces.metadata }) + .from(executionWorkspaces) + .where(eq(executionWorkspaces.id, seeded.executionWorkspaceId)); + + expect(sweep).toMatchObject({ archived: 0, skippedNonTerminalTree: 1 }); + expect(workspace?.status).toBe("active"); + expect(metadataHasReopenPendingConsumption(workspace?.metadata as Record | null)).toBe(false); + await expect(fs.access(seeded.worktreePath)).resolves.toBeUndefined(); + }); + + it("refuses to archive a reopen-pending workspace and leaves the row unchanged", async () => { + // Close the second destructive path. The archive route calls + // archiveWorkspaceUnderLifecycleLock. A reopen published this row active with + // the reopen-pending flag while the source issue is still terminal. The + // archive must not close or clear the flag, so the destruction fence never + // removes the rebuilt worktree during the reopen consumption window. + const seeded = await seedTerminalWorkspace({ mergedPr: true }); + await db + .update(executionWorkspaces) + .set({ + status: "active", + closedAt: null, + cleanupReason: null, + cleanupEligibleAt: null, + metadata: { + createdByRuntime: true, + [EXECUTION_WORKSPACE_LIFECYCLE_GENERATION_METADATA_KEY]: 4, + [EXECUTION_WORKSPACE_REOPEN_PENDING_METADATA_KEY]: true, + }, + }) + .where(eq(executionWorkspaces.id, seeded.executionWorkspaceId)); + + const result = await svc.archiveWorkspaceUnderLifecycleLock({ + id: seeded.executionWorkspaceId, + patch: {}, + closedAt: new Date(), + }); + + expect(result).toEqual({ outcome: "reopen_pending" }); + + const [workspace] = await db + .select({ + status: executionWorkspaces.status, + closedAt: executionWorkspaces.closedAt, + metadata: executionWorkspaces.metadata, + }) + .from(executionWorkspaces) + .where(eq(executionWorkspaces.id, seeded.executionWorkspaceId)); + + // The row stays active, keeps the flag, and keeps its generation. + expect(workspace?.status).toBe("active"); + expect(workspace?.closedAt).toBeNull(); + expect(metadataHasReopenPendingConsumption(workspace?.metadata as Record | null)).toBe(true); + expect( + (workspace?.metadata as Record | null)?.[EXECUTION_WORKSPACE_LIFECYCLE_GENERATION_METADATA_KEY], + ).toBe(4); + // The rebuilt worktree is intact. + await expect(fs.access(seeded.worktreePath)).resolves.toBeUndefined(); + }); + + it("does not overwrite a newer archive when a stale cleanup failure lands late", async () => { + // The archive route records a cleanup failure through the generation-fenced + // write after the destructive cleanup throws. Simulate a reopen and a fresh + // archive that raised the generation before the stale failure lands. The + // generation guard skips the stale write, so the newer archive keeps its own + // closedAt, cleanupReason, and status. + const seeded = await seedTerminalWorkspace({ mergedPr: true }); + const staleClosedAt = new Date(Date.now() - 60_000); + // The first archive closed the row at generation 2. + await db + .update(executionWorkspaces) + .set({ + status: "archived", + closedAt: staleClosedAt, + metadata: { + createdByRuntime: true, + [EXECUTION_WORKSPACE_LIFECYCLE_GENERATION_METADATA_KEY]: 2, + }, + }) + .where(eq(executionWorkspaces.id, seeded.executionWorkspaceId)); + + // A resume reopened the row and a fresh archive raised the generation to 3. + const newerClosedAt = new Date(); + await db + .update(executionWorkspaces) + .set({ + status: "archived", + closedAt: newerClosedAt, + cleanupReason: "newer archive", + metadata: { + createdByRuntime: true, + [EXECUTION_WORKSPACE_LIFECYCLE_GENERATION_METADATA_KEY]: 3, + }, + }) + .where(eq(executionWorkspaces.id, seeded.executionWorkspaceId)); + + // The first archive's cleanup failure lands late at the captured generation 2. + const skipped = await svc.applyClosedWorkspaceCleanupOutcome({ + id: seeded.executionWorkspaceId, + closedAt: staleClosedAt, + capturedGeneration: 2, + cleanupReason: "stale teardown boom", + markCleanupFailed: true, + }); + expect(skipped).toBeNull(); + + const [row] = await db + .select({ + status: executionWorkspaces.status, + closedAt: executionWorkspaces.closedAt, + cleanupReason: executionWorkspaces.cleanupReason, + }) + .from(executionWorkspaces) + .where(eq(executionWorkspaces.id, seeded.executionWorkspaceId)); + // The newer archive survives; the stale failure did not overwrite it. + expect(row?.status).toBe("archived"); + expect(row?.cleanupReason).toBe("newer archive"); + expect(row?.closedAt?.getTime()).toBe(newerClosedAt.getTime()); + }); + + it("applies the cleanup outcome only while the row is still closed at the captured generation", async () => { + // The archive route records the cleanup outcome after the destruction fence + // returns. While the row is still closed at the captured generation, the + // guarded write records the warnings and the cleanup_failed status. After a + // resume reopened the row and raised the generation, the guard skips the write + // so a stale patch does not overwrite the rebuilt worktree's active state. + const seeded = await seedTerminalWorkspace({ mergedPr: true }); + await db + .update(executionWorkspaces) + .set({ + status: "archived", + closedAt: new Date(), + metadata: { + createdByRuntime: true, + [EXECUTION_WORKSPACE_LIFECYCLE_GENERATION_METADATA_KEY]: 2, + }, + }) + .where(eq(executionWorkspaces.id, seeded.executionWorkspaceId)); + + const closedAt = new Date(); + const applied = await svc.applyClosedWorkspaceCleanupOutcome({ + id: seeded.executionWorkspaceId, + closedAt, + capturedGeneration: 2, + cleanupReason: "teardown warning", + markCleanupFailed: true, + }); + expect(applied?.status).toBe("cleanup_failed"); + expect(applied?.cleanupReason).toBe("teardown warning"); + + // Simulate a resume that reopened the row and raised the generation after the + // destruction fence returned. + await db + .update(executionWorkspaces) + .set({ + status: "active", + closedAt: null, + cleanupReason: null, + metadata: { + createdByRuntime: true, + [EXECUTION_WORKSPACE_LIFECYCLE_GENERATION_METADATA_KEY]: 3, + }, + }) + .where(eq(executionWorkspaces.id, seeded.executionWorkspaceId)); + + const skipped = await svc.applyClosedWorkspaceCleanupOutcome({ + id: seeded.executionWorkspaceId, + closedAt: new Date(), + capturedGeneration: 2, + cleanupReason: "stale teardown warning", + markCleanupFailed: true, + }); + expect(skipped).toBeNull(); + + const [row] = await db + .select({ status: executionWorkspaces.status, cleanupReason: executionWorkspaces.cleanupReason }) + .from(executionWorkspaces) + .where(eq(executionWorkspaces.id, seeded.executionWorkspaceId)); + // The reopened active row survives; the stale cleanup patch did not land. + expect(row?.status).toBe("active"); + expect(row?.cleanupReason).toBeNull(); + }); + + it("routes every terminal-workspace write through one generation-fenced gateway that skips a stale generation", async () => { + // One gateway gates every destructive terminal-workspace write. This test + // raises the lifecycle generation past the value each writer captured, then + // drives all four refactored writers. Each writer must skip, because the one + // gateway sees the newer generation. This proves the single choke-point. + + // Writer 1 (clearReopenPendingConsumptionUnderLock), reached through + // clearReopenPendingConsumptionForUnconsumedReopen. A reopen published the row + // active at generation 5 and set the flag. A newer reopen then raised the + // generation to 6. A clear that presents the stale generation 5 must skip. + const clearSeed = await seedTerminalWorkspace({ mergedPr: true }); + const clearSince = new Date(Date.now() - 60_000).toISOString(); + await db + .update(executionWorkspaces) + .set({ + status: "active", + closedAt: null, + cleanupReason: null, + cleanupEligibleAt: null, + metadata: { + createdByRuntime: true, + [EXECUTION_WORKSPACE_LIFECYCLE_GENERATION_METADATA_KEY]: 6, + [EXECUTION_WORKSPACE_REOPEN_PENDING_METADATA_KEY]: true, + [EXECUTION_WORKSPACE_REOPEN_PENDING_SINCE_METADATA_KEY]: clearSince, + }, + }) + .where(eq(executionWorkspaces.id, clearSeed.executionWorkspaceId)); + const clearResult = await svc.clearReopenPendingConsumptionForUnconsumedReopen({ + workspaceId: clearSeed.executionWorkspaceId, + issue: { id: clearSeed.sourceIssueId, companyId: clearSeed.companyId }, + actor: { agentId: null, actorType: "user" }, + expectedGeneration: 5, + }); + expect(clearResult).toEqual({ cleared: false }); + const [afterClear] = await db + .select({ metadata: executionWorkspaces.metadata }) + .from(executionWorkspaces) + .where(eq(executionWorkspaces.id, clearSeed.executionWorkspaceId)); + // The newer owner keeps its flag, so the stale clear did not touch the fence. + expect(metadataHasReopenPendingConsumption(afterClear?.metadata as Record | null)).toBe(true); + + // Writer 2 (refreshReopenPendingConsumptionUnderLock), reached through + // refreshReopenPendingConsumption. A refresh that presents the stale + // generation 5 must skip and leave the timestamp unchanged. + const refreshResult = await svc.refreshReopenPendingConsumption({ + workspaceId: clearSeed.executionWorkspaceId, + expectedGeneration: 5, + }); + expect(refreshResult).toEqual({ refreshed: false }); + const [afterRefresh] = await db + .select({ metadata: executionWorkspaces.metadata }) + .from(executionWorkspaces) + .where(eq(executionWorkspaces.id, clearSeed.executionWorkspaceId)); + expect( + readMetadataReopenPendingConsumptionSince(afterRefresh?.metadata as Record | null)?.toISOString(), + ).toBe(clearSince); + + // Writer 3 (cleanupTerminalWorkspace) runs its destructive cleanup through the + // same gateway call as fenceClosedWorkspaceDestruction, with the same + // closed-status guard. The row is closed at generation 3, but the caller + // captured generation 2, so the gateway skips and never runs the destroy body. + const destroySeed = await seedTerminalWorkspace({ mergedPr: true }); + await db + .update(executionWorkspaces) + .set({ + status: "archived", + closedAt: new Date(), + metadata: { + createdByRuntime: true, + [EXECUTION_WORKSPACE_LIFECYCLE_GENERATION_METADATA_KEY]: 3, + }, + }) + .where(eq(executionWorkspaces.id, destroySeed.executionWorkspaceId)); + const destroy = vi.fn(async () => "destroyed"); + const destroyResult = await svc.fenceClosedWorkspaceDestruction({ + workspaceId: destroySeed.executionWorkspaceId, + capturedGeneration: 2, + destroy, + }); + expect(destroyResult).toEqual({ skippedReopened: true }); + expect(destroy).not.toHaveBeenCalled(); + + // Writer 4 (markTerminalCleanupFailedFenced). The reaper archives the row at + // one generation and captures it. The cleanup then throws. Before the catch + // handler writes cleanup_failed, a reopen and a fresh archive raise the + // generation. The fenced write must skip, so the newer archive survives. + const failSeed = await seedTerminalWorkspace({ mergedPr: true }); + const newerReason = "newer_archive_lifecycle_marker"; + const racingService = executionWorkspaceService(db, { + resolvePullRequestDetails: async (_companyId, reference) => + pullRequestDetailsByKey.get(`${failSeed.companyId}:${reference.number}`) ?? { state: "unknown" }, + beforeTerminalWorkspaceCleanup: async (workspace) => { + await db + .update(executionWorkspaces) + .set({ + status: "archived", + cleanupReason: newerReason, + metadata: { [EXECUTION_WORKSPACE_LIFECYCLE_GENERATION_METADATA_KEY]: 2 }, + updatedAt: new Date(), + }) + .where(eq(executionWorkspaces.id, workspace.id)); + throw new Error("forced cleanup failure"); + }, + }); + const sweep = await racingService.sweepTerminalWorkspaces(); + expect(sweep).toMatchObject({ cleanupFailed: 1 }); + const [afterFail] = await db + .select({ + status: executionWorkspaces.status, + cleanupReason: executionWorkspaces.cleanupReason, + metadata: executionWorkspaces.metadata, + }) + .from(executionWorkspaces) + .where(eq(executionWorkspaces.id, failSeed.executionWorkspaceId)); + // The fenced write saw the raised generation and skipped, so the newer + // lifecycle state survives untouched. + expect(afterFail?.status).toBe("archived"); + expect(afterFail?.cleanupReason).toBe(newerReason); + expect( + (afterFail?.metadata as Record | null)?.[EXECUTION_WORKSPACE_LIFECYCLE_GENERATION_METADATA_KEY], + ).toBe(2); + }); + it("holds Git index and ref locks across terminal cleanup", async () => { const seeded = await seedTerminalWorkspace({ mergedPr: true }); await db.update(executionWorkspaces).set({ diff --git a/server/src/__tests__/issue-closed-workspace-routes.test.ts b/server/src/__tests__/issue-closed-workspace-routes.test.ts index d9c82cd7d0..378e5076b6 100644 --- a/server/src/__tests__/issue-closed-workspace-routes.test.ts +++ b/server/src/__tests__/issue-closed-workspace-routes.test.ts @@ -16,6 +16,8 @@ const mockIssueService = vi.hoisted(() => ({ const mockExecutionWorkspaceService = vi.hoisted(() => ({ getById: vi.fn(), + reopenClosedIsolatedExecutionWorkspaceForIssue: vi.fn(), + clearReopenPendingConsumptionForUnconsumedReopen: vi.fn(async () => ({ cleared: true })), })); const mockAccessService = vi.hoisted(() => ({ @@ -59,6 +61,7 @@ function registerServiceMocks() { vi.doMock("../services/execution-workspaces.js", () => ({ executionWorkspaceService: () => mockExecutionWorkspaceService, + STALE_REOPEN_PENDING_CONSUMPTION_GRACE_MS: 5 * 60 * 1000, })); vi.doMock("../services/heartbeat.js", () => ({ @@ -139,7 +142,7 @@ function registerServiceMocks() { })); } -async function createApp() { +async function createApp(actor?: Record) { const [{ issueRoutes }, { errorHandler }] = await Promise.all([ import("../routes/issues.js"), import("../middleware/index.js"), @@ -147,7 +150,7 @@ async function createApp() { const app = express(); app.use(express.json()); app.use((req, _res, next) => { - (req as any).actor = { + (req as any).actor = actor ?? { type: "board", userId: "local-board", companyIds: ["company-1"], @@ -208,30 +211,40 @@ describe.sequential("closed isolated workspace issue routes", () => { vi.clearAllMocks(); mockIssueService.getById.mockResolvedValue(makeIssue()); mockExecutionWorkspaceService.getById.mockResolvedValue(makeClosedWorkspace()); + // The guard reopens a closed isolated workspace and lets the request + // continue. The default is a successful reopen. + mockExecutionWorkspaceService.reopenClosedIsolatedExecutionWorkspaceForIssue.mockResolvedValue({ + ok: true, + reopened: true, + workspace: { ...makeClosedWorkspace(), status: "active", closedAt: null }, + generation: 4, + }); }); - it("rejects new issue comments when the linked isolated workspace is closed", async () => { + it("reopens the closed isolated workspace and accepts a new comment", async () => { const res = await request(await createApp()) .post(`/api/issues/${issueId}/comments`) .send({ body: "hello" }); - expect(res.status).toBe(409); - expect(res.body.error).toContain("closed workspace"); - expect(mockIssueService.addComment).not.toHaveBeenCalled(); + expect(mockExecutionWorkspaceService.reopenClosedIsolatedExecutionWorkspaceForIssue).toHaveBeenCalledWith({ + workspaceId: closedWorkspaceId, + issue: { id: issueId, companyId: "company-1", projectId: null }, + actor: expect.objectContaining({ actorType: "user" }), + }); + // The closed-workspace dead end is gone. + expect(res.status).not.toBe(409); }); - it("rejects comment updates when the linked isolated workspace is closed", async () => { + it("reopens the closed isolated workspace and accepts a comment update", async () => { const res = await request(await createApp()) .patch(`/api/issues/${issueId}`) .send({ comment: "hello" }); - expect(res.status).toBe(409); - expect(res.body.error).toContain("closed workspace"); - expect(mockIssueService.update).not.toHaveBeenCalled(); - expect(mockIssueService.addComment).not.toHaveBeenCalled(); + expect(mockExecutionWorkspaceService.reopenClosedIsolatedExecutionWorkspaceForIssue).toHaveBeenCalledTimes(1); + expect(res.status).not.toBe(409); }); - it("rejects checkout when the linked isolated workspace is closed", async () => { + it("reopens the closed isolated workspace and accepts a checkout", async () => { const res = await request(await createApp()) .post(`/api/issues/${issueId}/checkout`) .send({ @@ -239,11 +252,190 @@ describe.sequential("closed isolated workspace issue routes", () => { expectedStatuses: ["todo", "backlog", "blocked"], }); + expect(mockExecutionWorkspaceService.reopenClosedIsolatedExecutionWorkspaceForIssue).toHaveBeenCalledTimes(1); + expect(res.status).not.toBe(409); + }); + + it("returns 409 and blocks the comment when the workspace cannot be reopened", async () => { + mockExecutionWorkspaceService.reopenClosedIsolatedExecutionWorkspaceForIssue.mockResolvedValue({ + ok: false, + code: "not_reopenable", + message: "Execution workspace is not reopenable", + }); + + const res = await request(await createApp()) + .post(`/api/issues/${issueId}/comments`) + .send({ body: "hello" }); + expect(res.status).toBe(409); - expect(res.body.error).toContain("closed workspace"); + expect(mockIssueService.addComment).not.toHaveBeenCalled(); + }); + + it("returns 503 and blocks the checkout when the rebuild fails", async () => { + mockExecutionWorkspaceService.reopenClosedIsolatedExecutionWorkspaceForIssue.mockResolvedValue({ + ok: false, + code: "rebuild_failed", + message: "Failed to rebuild the execution workspace", + }); + + const res = await request(await createApp()) + .post(`/api/issues/${issueId}/checkout`) + .send({ + agentId, + expectedStatuses: ["todo", "backlog", "blocked"], + }); + + expect(res.status).toBe(503); expect(mockIssueService.checkout).not.toHaveBeenCalled(); }); + it("does not reopen the workspace when a checkout fails the run-id gate", async () => { + // An agent checkout without a run id is rejected before the reopen runs. The + // reopen must not rebuild and republish the workspace, or the still-terminal + // issue keeps a leaked active workspace that the reaper skips. + const agentActorWithoutRunId = { + type: "agent", + agentId, + companyId: "company-1", + runId: null, + source: "agent_key", + }; + + const res = await request(await createApp(agentActorWithoutRunId)) + .post(`/api/issues/${issueId}/checkout`) + .send({ + agentId, + expectedStatuses: ["todo", "backlog", "blocked"], + }); + + expect(res.status).toBe(401); + expect( + mockExecutionWorkspaceService.reopenClosedIsolatedExecutionWorkspaceForIssue, + ).not.toHaveBeenCalled(); + expect(mockIssueService.checkout).not.toHaveBeenCalled(); + }); + + it("clears the reopen-pending flag when the comment update returns null after a reopen", async () => { + // The workspace reopens, but the issue update then returns null. The issue + // stays terminal, so the guard must clear the reopen-pending flag. Otherwise + // the terminal reaper and the archive route skip the rebuilt worktree forever. + mockIssueService.getById.mockResolvedValue({ ...makeIssue(), status: "done" }); + mockIssueService.update.mockResolvedValue(null); + + const res = await request(await createApp()) + .patch(`/api/issues/${issueId}`) + .send({ comment: "hello" }); + + expect(res.status).toBe(404); + await vi.waitFor(() => { + expect( + mockExecutionWorkspaceService.clearReopenPendingConsumptionForUnconsumedReopen, + ).toHaveBeenCalledWith( + expect.objectContaining({ + workspaceId: closedWorkspaceId, + issue: expect.objectContaining({ id: issueId }), + expectedGeneration: 4, + }), + ); + }); + }); + + it("clears the reopen-pending flag when the checkout throws after a reopen", async () => { + mockIssueService.getById.mockResolvedValue({ ...makeIssue(), status: "done" }); + mockIssueService.checkout.mockRejectedValue(new Error("checkout failed")); + + const res = await request(await createApp()) + .post(`/api/issues/${issueId}/checkout`) + .send({ + agentId, + expectedStatuses: ["todo", "backlog", "blocked"], + }); + + expect(res.status).toBe(500); + await vi.waitFor(() => { + expect( + mockExecutionWorkspaceService.clearReopenPendingConsumptionForUnconsumedReopen, + ).toHaveBeenCalledWith( + expect.objectContaining({ + workspaceId: closedWorkspaceId, + issue: expect.objectContaining({ id: issueId }), + expectedGeneration: 4, + }), + ); + }); + }); + + it("does not clear the reopen-pending flag when the checkout resumes the issue", async () => { + // The checkout moves the issue out of the terminal state, so the reaper clears + // the flag on its next cycle. The route must not clear it here. + mockIssueService.getById.mockResolvedValue({ ...makeIssue(), status: "done" }); + mockIssueService.checkout.mockResolvedValue({ ...makeIssue(), status: "in_progress" }); + + const res = await request(await createApp()) + .post(`/api/issues/${issueId}/checkout`) + .send({ + agentId, + expectedStatuses: ["todo", "backlog", "blocked"], + }); + + expect(res.status).toBe(200); + await new Promise((resolve) => setImmediate(resolve)); + expect( + mockExecutionWorkspaceService.clearReopenPendingConsumptionForUnconsumedReopen, + ).not.toHaveBeenCalled(); + }); + + it("does not clear the reopen-pending flag when a concurrent request already reopened the workspace", async () => { + // A concurrent request reopened the workspace first, so this request receives + // reopened: false and never set the flag. Even though the checkout leaves the + // issue terminal, this request must not clear the flag that the other request + // owns. Otherwise the reaper or the archive route can destroy the rebuilt + // worktree while the other request still uses it. + mockIssueService.getById.mockResolvedValue({ ...makeIssue(), status: "done" }); + mockExecutionWorkspaceService.reopenClosedIsolatedExecutionWorkspaceForIssue.mockResolvedValue({ + ok: true, + reopened: false, + workspace: { ...makeClosedWorkspace(), status: "active", closedAt: null }, + generation: 4, + }); + mockIssueService.checkout.mockResolvedValue(null); + + const res = await request(await createApp()) + .post(`/api/issues/${issueId}/checkout`) + .send({ + agentId, + expectedStatuses: ["todo", "backlog", "blocked"], + }); + + expect(res.status).toBe(200); + await new Promise((resolve) => setImmediate(resolve)); + expect( + mockExecutionWorkspaceService.clearReopenPendingConsumptionForUnconsumedReopen, + ).not.toHaveBeenCalled(); + }); + + it("retries the reopen-pending clear when the first attempt fails transiently", async () => { + // The workspace reopens, but the comment update returns null, so the issue + // stays terminal and the guard must clear the flag. A transient failure of the + // first clear must not strand the flag; the guard retries until it succeeds. + mockIssueService.getById.mockResolvedValue({ ...makeIssue(), status: "done" }); + mockIssueService.update.mockResolvedValue(null); + mockExecutionWorkspaceService.clearReopenPendingConsumptionForUnconsumedReopen + .mockRejectedValueOnce(new Error("transient database error")) + .mockResolvedValue({ cleared: true }); + + const res = await request(await createApp()) + .patch(`/api/issues/${issueId}`) + .send({ comment: "hello" }); + + expect(res.status).toBe(404); + await vi.waitFor(() => { + expect( + mockExecutionWorkspaceService.clearReopenPendingConsumptionForUnconsumedReopen.mock.calls.length, + ).toBeGreaterThanOrEqual(2); + }); + }); + it("still allows non-comment board updates so the issue can be moved to a new workspace", async () => { mockIssueService.update.mockResolvedValue({ ...makeIssue(), diff --git a/server/src/__tests__/issue-feedback-routes.test.ts b/server/src/__tests__/issue-feedback-routes.test.ts index ff0ac6e98a..b84150ef2b 100644 --- a/server/src/__tests__/issue-feedback-routes.test.ts +++ b/server/src/__tests__/issue-feedback-routes.test.ts @@ -116,6 +116,7 @@ function registerModuleMocks() { vi.doMock("../services/execution-workspaces.js", () => ({ executionWorkspaceService: () => mockExecutionWorkspaceService, + STALE_REOPEN_PENDING_CONSUMPTION_GRACE_MS: 5 * 60 * 1000, })); vi.doMock("../services/feedback.js", () => ({ diff --git a/server/src/__tests__/issue-workspace-command-authz.test.ts b/server/src/__tests__/issue-workspace-command-authz.test.ts index a79a9de8c1..238419dbe1 100644 --- a/server/src/__tests__/issue-workspace-command-authz.test.ts +++ b/server/src/__tests__/issue-workspace-command-authz.test.ts @@ -81,6 +81,7 @@ function registerRouteMocks() { vi.doMock("../services/execution-workspaces.js", () => ({ executionWorkspaceService: () => mockExecutionWorkspaceService, + STALE_REOPEN_PENDING_CONSUMPTION_GRACE_MS: 5 * 60 * 1000, })); vi.doMock("../services/feedback.js", () => ({ diff --git a/server/src/__tests__/issues-goal-context-routes.test.ts b/server/src/__tests__/issues-goal-context-routes.test.ts index affac46c00..55a064ebcd 100644 --- a/server/src/__tests__/issues-goal-context-routes.test.ts +++ b/server/src/__tests__/issues-goal-context-routes.test.ts @@ -137,6 +137,7 @@ vi.mock("../services/index.js", () => ({ vi.mock("../services/execution-workspaces.js", () => ({ executionWorkspaceService: () => mockExecutionWorkspaceService, + STALE_REOPEN_PENDING_CONSUMPTION_GRACE_MS: 5 * 60 * 1000, })); function createApp() { diff --git a/server/src/routes/execution-workspaces.ts b/server/src/routes/execution-workspaces.ts index 2d4b592dba..77f13d4eab 100644 --- a/server/src/routes/execution-workspaces.ts +++ b/server/src/routes/execution-workspaces.ts @@ -13,7 +13,10 @@ import { import type { WorkspaceRuntimeDesiredState, WorkspaceRuntimeServiceStateMap } from "@paperclipai/shared"; import { validate } from "../middleware/validate.js"; import { accessService, executionWorkspaceService, heartbeatService, logActivity, workspaceOperationService } from "../services/index.js"; -import { mergeExecutionWorkspaceConfig, readExecutionWorkspaceConfig } from "../services/execution-workspaces.js"; +import { + mergeExecutionWorkspaceConfig, + readExecutionWorkspaceConfig, +} from "../services/execution-workspaces.js"; import { parseProjectExecutionWorkspacePolicy } from "../services/execution-workspace-policy.js"; import { readProjectWorkspaceRuntimeConfig } from "../services/project-workspace-runtime-config.js"; import { @@ -634,23 +637,33 @@ export function executionWorkspaceRoutes(db: Db, opts: { pluginWorkerManager?: P } const closedAt = new Date(); - const archivedWorkspace = await svc.update(id, { - ...patch, - status: "archived", + // Archive under the per-workspace lifecycle lock. The service takes the same + // lock as a reopen, raises the lifecycle generation, and clears the + // reopen-pending flag. The lock stops a concurrent reopen from publishing an + // active row between the status re-check and this archive write, so the + // destruction fence below never deletes a worktree that a reopen rebuilt. + const archiveResult = await svc.archiveWorkspaceUnderLifecycleLock({ + id, + patch, closedAt, - cleanupReason: null, }); - if (!archivedWorkspace) { + if (!archiveResult) { res.status(404).json({ error: "Execution workspace not found" }); return; } - workspace = archivedWorkspace; - - await environmentRuntime.destroyReusableSandboxLeases({ - companyId: existing.companyId, - executionWorkspaceId: existing.id, - failureReason: "execution_workspace_closed", - }); + if (archiveResult.outcome === "reopen_pending") { + // A reopen published this workspace as active while its source issue is + // still terminal. A caller will consume the rebuilt worktree. Refuse the + // archive and return before any lease teardown, runtime-service stop, or + // artifact cleanup, so the archive control never removes the rebuilt + // worktree during the reopen consumption window. + res.status(409).json({ + error: "Execution workspace was reopened and cannot be archived right now", + }); + return; + } + workspace = archiveResult.workspace; + const capturedGeneration = archiveResult.capturedGeneration; if (existing.mode === "shared_workspace") { await db @@ -668,11 +681,6 @@ export function executionWorkspaceRoutes(db: Db, opts: { pluginWorkerManager?: P } try { - await stopRuntimeServicesForExecutionWorkspace({ - db, - executionWorkspaceId: existing.id, - workspaceCwd: existing.cwd, - }); const projectWorkspace = existing.projectWorkspaceId ? await db .select({ @@ -697,35 +705,83 @@ export function executionWorkspaceRoutes(db: Db, opts: { pluginWorkerManager?: P .where(and(eq(projects.id, existing.projectId), eq(projects.companyId, existing.companyId))) .then((rows) => parseProjectExecutionWorkspacePolicy(rows[0]?.executionWorkspacePolicy)) : null; - const cleanupResult = await cleanupExecutionWorkspaceArtifacts({ - workspace: existing, - projectWorkspace, - teardownCommand: configForCleanup?.teardownCommand ?? projectPolicy?.workspaceStrategy?.teardownCommand ?? null, - cleanupCommand: configForCleanup?.cleanupCommand ?? null, - recorder: workspaceOperationsSvc.createRecorder({ - companyId: existing.companyId, - executionWorkspaceId: existing.id, - }), + // Destroy under the lifecycle lock. If a resume reopened the workspace in + // the meantime, the fence skips destruction and keeps the reopened row. + // The reusable sandbox lease teardown runs inside this fence too. A reopen + // that races the archive rebuilds the worktree and keeps its leases, so + // the fence must skip both the worktree teardown and the lease teardown at + // the same generation. If the lease teardown ran before the fence, an + // overlapping reopen would lose its reusable leases while the fence still + // preserved its rebuilt worktree. + const fenced = await svc.fenceClosedWorkspaceDestruction({ + workspaceId: id, + capturedGeneration, + destroy: async () => { + await environmentRuntime.destroyReusableSandboxLeases({ + companyId: existing.companyId, + executionWorkspaceId: existing.id, + failureReason: "execution_workspace_closed", + }); + await stopRuntimeServicesForExecutionWorkspace({ + db, + executionWorkspaceId: existing.id, + workspaceCwd: existing.cwd, + }); + return cleanupExecutionWorkspaceArtifacts({ + workspace: existing, + projectWorkspace, + teardownCommand: configForCleanup?.teardownCommand ?? projectPolicy?.workspaceStrategy?.teardownCommand ?? null, + cleanupCommand: configForCleanup?.cleanupCommand ?? null, + recorder: workspaceOperationsSvc.createRecorder({ + companyId: existing.companyId, + executionWorkspaceId: existing.id, + }), + }); + }, }); - cleanupWarnings = cleanupResult.warnings; - const cleanupPatch: Record = { - closedAt, - cleanupReason: cleanupWarnings.length > 0 ? cleanupWarnings.join(" | ") : null, - }; - if (!cleanupResult.cleaned) { - cleanupPatch.status = "cleanup_failed"; - } - if (cleanupResult.warnings.length > 0 || !cleanupResult.cleaned) { - workspace = (await svc.update(id, cleanupPatch)) ?? workspace; + if (fenced.skippedReopened) { + // A resume reopened the workspace. Return the current (active) row. + workspace = (await svc.getById(id)) ?? workspace; + } else { + const cleanupResult = fenced.result; + cleanupWarnings = cleanupResult.warnings; + if (cleanupResult.warnings.length > 0 || !cleanupResult.cleaned) { + // Record the cleanup outcome under the lifecycle lock at the captured + // generation. If a resume reopened the workspace after the destruction + // fence returned, the guarded write skips the row, so a stale patch + // never overwrites the rebuilt worktree's active state. + const applied = await svc.applyClosedWorkspaceCleanupOutcome({ + id, + closedAt, + capturedGeneration, + cleanupReason: cleanupWarnings.length > 0 ? cleanupWarnings.join(" | ") : null, + markCleanupFailed: !cleanupResult.cleaned, + }); + if (applied) { + workspace = applied; + } else { + // A resume reopened the workspace. Return the current (active) row. + workspace = (await svc.getById(id)) ?? workspace; + } + } } } catch (error) { const failureReason = error instanceof Error ? error.message : String(error); - workspace = - (await svc.update(id, { - status: "cleanup_failed", - closedAt, - cleanupReason: failureReason, - })) ?? workspace; + // Mark cleanup_failed only while the row is still closed at the captured + // generation. If a resume reopened the workspace after the cleanup threw, + // the row is active again and, after a fresh archive, carries a higher + // generation. The generation-fenced write skips the row in both cases, so + // a stale cleanup_failed write never overwrites the newer active lifecycle + // state, and never buries a newer archive under the first archive's + // failure. + const marked = await svc.applyClosedWorkspaceCleanupOutcome({ + id, + closedAt, + capturedGeneration, + cleanupReason: failureReason, + markCleanupFailed: true, + }); + if (marked) workspace = marked; res.status(500).json({ error: `Failed to archive execution workspace: ${failureReason}`, }); diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index dbdfe47064..f2030255d5 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -66,7 +66,6 @@ import { updateDocumentAnnotationThreadSchema, upsertIssueDocumentSchema, updateIssueSchema, - getClosedIsolatedExecutionWorkspaceMessage, isClosedIsolatedExecutionWorkspace, isUuidLike, normalizeIssueIdentifier as normalizeIssueReferenceIdentifier, @@ -179,7 +178,10 @@ import { findExistingIssueBlockersResolvedWake, } from "../services/issue-dependency-wakeups.js"; import { assertEnvironmentSelectionForCompany } from "./environment-selection.js"; -import { executionWorkspaceService as executionWorkspaceServiceDirect } from "../services/execution-workspaces.js"; +import { + executionWorkspaceService as executionWorkspaceServiceDirect, + STALE_REOPEN_PENDING_CONSUMPTION_GRACE_MS, +} from "../services/execution-workspaces.js"; import { decisionTrainingService } from "../services/decision-training.js"; import { feedbackService } from "../services/feedback.js"; import { instanceSettingsService } from "../services/instance-settings.js"; @@ -5018,14 +5020,159 @@ export function issueRoutes( return workspace; } - function respondClosedIssueExecutionWorkspace( + // Reopen the closed isolated workspace that a guard found, so the request can + // continue. The return value tells the caller what happened: + // "reopened" - this request rebuilt the workspace and set the + // reopen-pending flag. The caller must install the + // consumption guard so the flag cannot leak. + // "already-open" - a concurrent request already reopened the workspace, so + // this request did not set the flag. The caller continues but + // must not install the guard, or it can clear the flag that + // the other request still owns. + // null - this function sent an error response, so the caller stops. + // The reopen is scoped to the issue company and project inside the service, and + // it runs only after the route already authorized the request on the issue. + async function reopenClosedIssueExecutionWorkspaceOrRespond( + req: Request, res: Response, - workspace: Pick, - ) { - res.status(409).json({ - error: getClosedIsolatedExecutionWorkspaceMessage(workspace), - executionWorkspace: workspace, + issue: { id: string; companyId: string; projectId?: string | null }, + workspace: Pick, + ): Promise<{ outcome: "reopened" | "already-open"; generation: number } | null> { + const actor = getActorInfo(req); + const result = await executionWorkspacesSvc.reopenClosedIsolatedExecutionWorkspaceForIssue({ + workspaceId: workspace.id, + issue: { id: issue.id, companyId: issue.companyId, projectId: issue.projectId ?? null }, + actor: { agentId: actor.agentId, actorType: actor.actorType }, }); + if (result.ok) { + return { outcome: result.reopened ? "reopened" : "already-open", generation: result.generation }; + } + if (result.code === "not_reopenable") { + res.status(409).json({ error: "This issue is linked to a closed workspace that cannot be reopened." }); + } else { + res.status(503).json({ error: "Could not reopen the workspace for this issue. Please try again." }); + } + return null; + } + + // The keepalive re-stamps the reopen-pending flag on this interval while a + // consuming request is in flight. The interval is one fifth of the stale grace + // period, so several re-stamps land before the reaper could treat the flag as + // stranded. This keeps a live but slow request's fence against the reaper. + const REOPEN_PENDING_REFRESH_INTERVAL_MS = Math.floor( + STALE_REOPEN_PENDING_CONSUMPTION_GRACE_MS / 5, + ); + + // Guard a reopen against a caller that never consumes it. + // `reopenClosedIssueExecutionWorkspaceOrRespond` publishes the rebuilt worktree + // as active and sets the reopen-pending flag while the source issue is still + // terminal. The route then moves the issue out of the terminal state, and the + // terminal reaper clears the flag once it sees the non-terminal issue. If the + // route mutation returns null, throws, or leaves the issue terminal, the flag + // stays set and both the reaper and the archive route skip the row forever, so + // the rebuilt worktree leaks and no path can reclaim it. + // + // This guard runs when the response ends, so it covers every exit: a success, a + // rejected mutation, and a thrown error. It reads the final issue status through + // a getter. When the issue is null or still terminal, it clears the flag so the + // reaper can reclaim the worktree. When the issue left the terminal state, it + // does nothing and the reaper clears the flag. The guard never touches the + // response, and the underlying clear is idempotent. + function guardReopenedWorkspaceConsumption(input: { + req: Request; + res: Response; + issue: { id: string; companyId: string }; + workspace: Pick | null; + generation: number | null; + finalIssueStatus: () => string | null | undefined; + }): void { + const { req, res, issue, workspace, generation, finalIssueStatus } = input; + if (!workspace || generation === null) return; + // Re-stamp the reopen-pending flag while this request is in flight. The + // request that consumes the rebuilt worktree is an HTTP request, not a + // heartbeat run, so the terminal reaper cannot see it through + // `workspaceHasActiveRun`. A request that outruns the stale grace period + // would let the reaper clear the live fence, and a later sweep would archive + // and destroy the worktree under the request. The keepalive re-stamps the + // timestamp on an interval below the grace, so the flag never looks stranded + // while the request lives. The refresh runs only while the flag is still set + // and the generation still matches, so it never revives a cleared flag and + // never refreshes a newer reopen's fence. + const keepAlive = setInterval(() => { + void executionWorkspacesSvc + .refreshReopenPendingConsumption({ + workspaceId: workspace.id, + expectedGeneration: generation, + }) + .then((result) => { + // The fence is no longer ours: a clear removed the flag, or a newer + // reopen or an archive raised the generation. Stop the keepalive so it + // does not re-stamp another owner's row. + if (!result.refreshed) clearInterval(keepAlive); + }) + .catch((err) => { + // A transient database error must not stop the keepalive. Keep the + // interval so the next tick retries before the grace period elapses. + logger.warn( + { err, issueId: issue.id, executionWorkspaceId: workspace.id }, + "failed to refresh the reopen-pending flag for an in-flight request", + ); + }); + }, REOPEN_PENDING_REFRESH_INTERVAL_MS); + // Do not keep the event loop alive for the keepalive alone. + keepAlive.unref?.(); + let settled = false; + const settle = () => { + if (settled) return; + settled = true; + clearInterval(keepAlive); + const status = finalIssueStatus(); + if (typeof status === "string" && !isClosedIssueStatus(status)) return; + const actor = getActorInfo(req); + void clearReopenPendingConsumptionWithRetry({ + workspaceId: workspace.id, + issue: { id: issue.id, companyId: issue.companyId }, + actor: { agentId: actor.agentId, actorType: actor.actorType }, + expectedGeneration: generation, + }); + }; + res.once("finish", settle); + res.once("close", settle); + } + + // Clear the reopen-pending flag with a bounded retry. The response already + // ended when this runs, so it is a background best-effort. A transient database + // error must not strand the flag: while the flag stays set, the terminal reaper + // skips the workspace and the archive route rejects it, so the rebuilt worktree + // leaks. The clear is idempotent, so a retry after a partial failure is safe. + // The method returns { cleared: false } without an error when the flag is + // already clear, so that path does not retry. + async function clearReopenPendingConsumptionWithRetry(input: { + workspaceId: string; + issue: { id: string; companyId: string }; + actor: { agentId: string | null; actorType: string }; + expectedGeneration: number; + }): Promise { + const maxAttempts = 5; + for (let attempt = 1; attempt <= maxAttempts; attempt += 1) { + try { + await executionWorkspacesSvc.clearReopenPendingConsumptionForUnconsumedReopen(input); + return; + } catch (err) { + if (attempt >= maxAttempts) { + logger.error( + { err, issueId: input.issue.id, executionWorkspaceId: input.workspaceId, attempts: attempt }, + "failed to clear the reopen-pending flag after an unconsumed reopen; the rebuilt worktree may leak until the flag clears", + ); + return; + } + logger.warn( + { err, issueId: input.issue.id, executionWorkspaceId: input.workspaceId, attempt }, + "retry the clear of the reopen-pending flag after an unconsumed reopen", + ); + await new Promise((resolve) => setTimeout(resolve, attempt * 250)); + } + } } async function destroyReusableSandboxLeasesForTerminalIssue(issue: { @@ -8737,10 +8884,6 @@ export function issueRoutes( req.actor.type === "agent" && (Object.keys(updateFields).length > 0 || reviewRequest !== undefined || hiddenAtRaw !== undefined); - if (closedExecutionWorkspace && (commentBody || isAgentWorkUpdate)) { - respondClosedIssueExecutionWorkspace(res, closedExecutionWorkspace); - return; - } if ( isAgentWorkUpdate && !(await assertCrossIssueInfluenceWithinRunCap(req, res, existing, "update")) @@ -9066,7 +9209,43 @@ export function issueRoutes( }, }, postCommitActivityPublications); }; + // Reopen the closed isolated workspace only after every access, validation, + // and policy gate passes, and just before the update persists. A rejected + // update must not rebuild and republish the workspace as active, because the + // issue stays terminal and the reaper then skips the leaked workspace. + let reopenedWorkspace: Pick | null = null; + let reopenedGeneration: number | null = null; + if (closedExecutionWorkspace && (commentBody || isAgentWorkUpdate)) { + const reopenOutcome = await reopenClosedIssueExecutionWorkspaceOrRespond( + req, + res, + existing, + closedExecutionWorkspace, + ); + if (reopenOutcome === null) { + return; + } + // Install the guard only when this request set the reopen-pending flag. A + // concurrent request that found the workspace already open must not clear + // the flag that the actual reopener still owns. + if (reopenOutcome.outcome === "reopened") { + reopenedWorkspace = closedExecutionWorkspace; + reopenedGeneration = reopenOutcome.generation; + } + } let issue: Awaited>; + // Clear the reopen-pending flag if this update leaves the issue terminal, so + // the rebuilt worktree does not leak. The guard reads `issue` when the + // response ends, so it also covers a null return and a thrown error. It clears + // only the fence this request installed, keyed by its generation. + guardReopenedWorkspaceConsumption({ + req, + res, + issue: existing, + workspace: reopenedWorkspace, + generation: reopenedGeneration, + finalIssueStatus: () => issue?.status, + }); try { if (transition.decision && decisionId) { const decision = transition.decision; @@ -10119,14 +10298,45 @@ export function issueRoutes( } const closedExecutionWorkspace = await getClosedIssueExecutionWorkspace(issue); - if (closedExecutionWorkspace) { - respondClosedIssueExecutionWorkspace(res, closedExecutionWorkspace); - return; - } const checkoutRunId = requireAgentRunId(req, res); if (req.actor.type === "agent" && !checkoutRunId) return; - let updated; + + // Reopen the closed isolated workspace only after the run-id gate passes. A + // rejected checkout must not rebuild and republish the workspace as active. + let reopenedWorkspace: Pick | null = null; + let reopenedGeneration: number | null = null; + if (closedExecutionWorkspace) { + const reopenOutcome = await reopenClosedIssueExecutionWorkspaceOrRespond( + req, + res, + issue, + closedExecutionWorkspace, + ); + if (reopenOutcome === null) { + return; + } + // Install the guard only when this request set the reopen-pending flag. A + // concurrent request that found the workspace already open must not clear + // the flag that the actual reopener still owns. + if (reopenOutcome.outcome === "reopened") { + reopenedWorkspace = closedExecutionWorkspace; + reopenedGeneration = reopenOutcome.generation; + } + } + let updated: Awaited> | undefined; + // Clear the reopen-pending flag if the checkout leaves the issue terminal, so + // the rebuilt worktree does not leak. The guard reads `updated` when the + // response ends, so it covers a null return and a thrown error. It clears only + // the fence this request installed, keyed by its generation. + guardReopenedWorkspaceConsumption({ + req, + res, + issue, + workspace: reopenedWorkspace, + generation: reopenedGeneration, + finalIssueStatus: () => updated?.status, + }); try { updated = await svc.checkout(id, req.body.agentId, req.body.expectedStatuses, checkoutRunId); } catch (error) { @@ -11109,10 +11319,6 @@ export function issueRoutes( metadata: req.body.metadata, })) return; const closedExecutionWorkspace = await getClosedIssueExecutionWorkspace(issue); - if (closedExecutionWorkspace) { - respondClosedIssueExecutionWorkspace(res, closedExecutionWorkspace); - return; - } const actor = getActorInfo(req); const commentPresentation = req.body.presentation ?? @@ -11193,10 +11399,48 @@ export function issueRoutes( return; } if (!(await assertCrossIssueInfluenceWithinRunCap(req, res, issue, "comment"))) return; + // Reopen the closed isolated workspace only after every access, resume-intent, + // blocker, and run-cap gate passes. A rejected comment must not rebuild and + // republish the workspace as active, because the issue stays terminal and the + // reaper then skips the leaked workspace. + let reopenedWorkspace: Pick | null = null; + let reopenedGeneration: number | null = null; + if (closedExecutionWorkspace) { + const reopenOutcome = await reopenClosedIssueExecutionWorkspaceOrRespond( + req, + res, + issue, + closedExecutionWorkspace, + ); + if (reopenOutcome === null) { + return; + } + // Install the guard only when this request set the reopen-pending flag. A + // concurrent request that found the workspace already open must not clear + // the flag that the actual reopener still owns. + if (reopenOutcome.outcome === "reopened") { + reopenedWorkspace = closedExecutionWorkspace; + reopenedGeneration = reopenOutcome.generation; + } + } let reopened = false; let reopenFromStatus: string | null = null; let interruptedRunId: string | null = null; let currentIssue = issue; + // Clear the reopen-pending flag if this comment leaves the issue terminal, so + // the rebuilt worktree does not leak. A comment reopens the workspace but only + // moves the issue out of the terminal state when it resumes the work. The + // guard reads `currentIssue` when the response ends, so it covers a rejected + // move, a thrown error, and a comment that keeps the issue terminal. It clears + // only the fence this request installed, keyed by its generation. + guardReopenedWorkspaceConsumption({ + req, + res, + issue, + workspace: reopenedWorkspace, + generation: reopenedGeneration, + finalIssueStatus: () => currentIssue.status, + }); let issueBeforeCommentDecision = issue; let commentDecisionStageWakeup: ReturnType | null = null; const commentReferenceSummaryBefore = await issueReferencesSvc.listIssueReferenceSummary(issue.id); diff --git a/server/src/services/execution-workspaces.ts b/server/src/services/execution-workspaces.ts index 37b5d438b4..4e32a69a37 100644 --- a/server/src/services/execution-workspaces.ts +++ b/server/src/services/execution-workspaces.ts @@ -35,6 +35,7 @@ import type { } from "@paperclipai/shared"; import { deriveProjectUrlKey, WORKSPACE_OVERVIEW_LINKED_ISSUE_LIMIT } from "@paperclipai/shared"; import { conflict, notFound, unprocessable } from "../errors.js"; +import { logger } from "../middleware/logger.js"; import { applyIssueExecutionPolicyTransition, normalizeIssueExecutionPolicy, @@ -61,12 +62,146 @@ import { type ExecutionWorkspaceRow = typeof executionWorkspaces.$inferSelect; type WorkspaceRuntimeServiceRow = typeof workspaceRuntimeServices.$inferSelect; type RuntimeServiceReadDb = Pick; +type DbTransaction = Parameters[0]>[0]; const execFileAsync = promisify(execFile); const TERMINAL_ISSUE_STATUSES = new Set(["done", "cancelled"]); const WORKSPACE_BRANCH_INCOHERENCE_REASON = "git_worktree_branch_incoherence"; const WORKSPACE_VALIDATION_RECOVERY_CAUSE = "workspace_validation_failed"; export const ISSUE_TERMINAL_WORKSPACE_CLEANUP_REASON = "issue_terminal"; +// The reopen-failure reason kept on the row when a rebuild does not finish. The +// value is sanitized: it never contains a repository URL, a host path, or git +// output. +export const EXECUTION_WORKSPACE_REOPEN_FAILED_REASON = "reopen_failed"; + +// How long the terminal reaper waits before it reclaims a stranded reopen-pending +// flag. A reopen sets the flag while the source issue is still terminal. The +// consuming request clears the flag within seconds when the request ends. If the +// server exits first, or every clear retry fails, the flag stays set and both the +// reaper and the archive route skip the workspace forever. After this grace the +// reaper reclaims the flag, but only when no run still owns it. The reaper checks +// for a live consuming run first, so a request that outruns this grace keeps its +// fence. A run has a heartbeat row the reaper can see. An HTTP consuming request +// has none, so the route re-stamps the flag on an interval below this grace, and +// the fresh timestamp keeps the fence. The grace is a backstop for a flag whose +// consumer is gone, not a hard deadline on the request. +export const STALE_REOPEN_PENDING_CONSUMPTION_GRACE_MS = 5 * 60 * 1000; + +// The metadata key that holds the workspace lifecycle generation. The generation +// is a monotonic integer. Every archive and every reopen increases it by one. A +// destructive cleanup captures the generation it archived at, then re-reads the +// generation under the lifecycle lock immediately before it deletes the worktree. +// A reopen that ran in between raises the generation, so the stale cleanup finds +// a mismatch and does nothing. This fences a queued or in-flight cleanup against +// a workspace that a reopen already restored. +export const EXECUTION_WORKSPACE_LIFECYCLE_GENERATION_METADATA_KEY = "lifecycleGeneration"; + +export function readExecutionWorkspaceLifecycleGeneration( + metadata: Record | null | undefined, +): number { + const raw = metadata?.[EXECUTION_WORKSPACE_LIFECYCLE_GENERATION_METADATA_KEY]; + return typeof raw === "number" && Number.isInteger(raw) && raw >= 0 ? raw : 0; +} + +// Return a metadata object with the lifecycle generation increased by one. The +// caller keeps every other metadata key. +export function bumpExecutionWorkspaceLifecycleGeneration( + metadata: Record | null | undefined, +): Record { + return { + ...(metadata ?? {}), + [EXECUTION_WORKSPACE_LIFECYCLE_GENERATION_METADATA_KEY]: + readExecutionWorkspaceLifecycleGeneration(metadata) + 1, + }; +} + +function isClosedExecutionWorkspaceStatus(status: string | null | undefined): boolean { + return status === "archived" || status === "cleanup_failed"; +} + +// The metadata key that marks a workspace as reopened for a source issue that is +// still terminal. A reopen sets this flag in the same write that publishes the +// row as active. The source issue is still terminal at that moment, because the +// route changes the issue out of the terminal state only after the reopen +// returns. Both destructive paths (the terminal reaper and the archive route) +// exclude a flagged workspace, so neither one archives and destroys a worktree +// that a reopen rebuilt while the caller has not yet consumed it. The reaper +// clears the flag on a later pass once the source issue leaves the terminal +// state, so a normal terminal cycle can reap the workspace again. +export const EXECUTION_WORKSPACE_REOPEN_PENDING_METADATA_KEY = "reopenPendingConsumption"; + +// The metadata key that holds the time a reopen set the reopen-pending flag. The +// terminal reaper reads this timestamp to tell a fresh, in-flight reopen from a +// stranded flag whose consumer never cleared it. A reopen writes this key in the +// same write that sets the flag, and every clear removes both keys together. +export const EXECUTION_WORKSPACE_REOPEN_PENDING_SINCE_METADATA_KEY = "reopenPendingConsumptionSince"; + +export function metadataHasReopenPendingConsumption( + metadata: Record | null | undefined, +): boolean { + return metadata?.[EXECUTION_WORKSPACE_REOPEN_PENDING_METADATA_KEY] === true; +} + +// Read the time a reopen set the reopen-pending flag. Return null when the flag +// carries no valid timestamp. The terminal reaper uses this to tell a fresh, +// in-flight reopen from a stranded flag whose consumer never cleared it. +export function readMetadataReopenPendingConsumptionSince( + metadata: Record | null | undefined, +): Date | null { + const raw = metadata?.[EXECUTION_WORKSPACE_REOPEN_PENDING_SINCE_METADATA_KEY]; + if (typeof raw !== "string") return null; + const parsed = new Date(raw); + return Number.isNaN(parsed.getTime()) ? null : parsed; +} + +// Return a metadata object with the reopen-pending flag set. The caller keeps +// every other metadata key. The `at` timestamp records when the reopen set the +// flag, so the terminal reaper can reclaim a stranded flag after a grace period. +export function setMetadataReopenPendingConsumption( + metadata: Record | null | undefined, + at: Date, +): Record { + return { + ...(metadata ?? {}), + [EXECUTION_WORKSPACE_REOPEN_PENDING_METADATA_KEY]: true, + [EXECUTION_WORKSPACE_REOPEN_PENDING_SINCE_METADATA_KEY]: at.toISOString(), + }; +} + +// Return a metadata object with the reopen-pending flag removed. The caller +// keeps every other metadata key. The function removes the flag and its +// timestamp together, so no orphan timestamp survives a clear. +export function clearMetadataReopenPendingConsumption( + metadata: Record | null | undefined, +): Record { + const next = { ...(metadata ?? {}) }; + delete next[EXECUTION_WORKSPACE_REOPEN_PENDING_METADATA_KEY]; + delete next[EXECUTION_WORKSPACE_REOPEN_PENDING_SINCE_METADATA_KEY]; + return next; +} + +// Acquire the per-workspace, transaction-scoped Postgres advisory lock. Postgres +// releases the lock when the transaction that holds `tx` commits or rolls back. +// Both the reopen path and the destructive cleanup path acquire the same lock, +// so they never run against the same workspace at the same time, even on +// different server processes. +async function acquireExecutionWorkspaceLifecycleLock( + tx: DbTransaction, + workspaceId: string, +): Promise { + await tx.execute( + sql`select pg_advisory_xact_lock(hashtextextended(${`execution_workspace_lifecycle:${workspaceId}`}, 0))`, + ); +} + +export type ReopenClosedIsolatedExecutionWorkspaceResult = + // `generation` is the lifecycle generation the reopen published the active row + // at. It owns the reopen-pending flag. A later clear must present this same + // generation, so a stale actor never clears a newer reopen's fence. + | { ok: true; workspace: ExecutionWorkspace; reopened: true; generation: number } + | { ok: true; workspace: ExecutionWorkspace; reopened: false; generation: number } + | { ok: false; code: "not_reopenable" | "rebuild_failed"; message: string }; + export type ExecutionWorkspaceServiceOptions = { resolvePullRequestDetails?: PullRequestMergeDetailsResolver; now?: () => Date; @@ -1279,7 +1414,171 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic return active.length > 0; } - async function cleanupTerminalWorkspace(workspace: ExecutionWorkspaceRow, expectedHeadSha: string | null) { + type FenceableWorkspaceRow = { + status: string; + metadata: Record | null; + }; + + // The single generation-fenced gateway for a terminal-workspace write. It owns + // the whole guarded step. It acquires the per-workspace lifecycle lock, it + // re-reads the fresh row, and it compares the current lifecycle generation to + // the generation the caller captured. It runs the caller's write body only + // when the current generation still equals `expectedGeneration` and the fresh + // row still passes the caller's `isWriteTarget` guard. Otherwise it emits the + // optional skip log and returns the caller's skip value. Every fenced + // terminal-workspace write routes through this function, so no other function + // re-derives the lock, the fresh read, or the generation compare. + async function fenceLifecycleGenerationWrite(input: { + workspaceId: string; + expectedGeneration: number; + isWriteTarget: (fresh: FenceableWorkspaceRow) => boolean; + onSkip: () => T; + skipLog?: { event: string; message: string }; + write: (context: { tx: DbTransaction; fresh: FenceableWorkspaceRow }) => Promise; + }): Promise { + return db.transaction(async (tx) => { + await acquireExecutionWorkspaceLifecycleLock(tx, input.workspaceId); + const row = await tx + .select({ status: executionWorkspaces.status, metadata: executionWorkspaces.metadata }) + .from(executionWorkspaces) + .where(eq(executionWorkspaces.id, input.workspaceId)) + .then((rows) => rows[0] ?? null); + const fresh: FenceableWorkspaceRow | null = row + ? { status: row.status, metadata: row.metadata as Record | null } + : null; + const currentGeneration = fresh ? readExecutionWorkspaceLifecycleGeneration(fresh.metadata) : null; + if ( + !fresh + || currentGeneration !== input.expectedGeneration + || !input.isWriteTarget(fresh) + ) { + if (input.skipLog) { + logger.info( + { + event: input.skipLog.event, + reason: "reopened", + executionWorkspaceId: input.workspaceId, + capturedGeneration: input.expectedGeneration, + currentGeneration, + currentStatus: fresh?.status ?? null, + }, + input.skipLog.message, + ); + } + return input.onSkip(); + } + return input.write({ tx, fresh }); + }); + } + + // Clear the reopen-pending flag through the lifecycle gateway. Every caller + // presents the lifecycle generation that owns the fence it wants to clear. The + // gateway removes the flag only when the current generation still equals + // `expectedGeneration`. A newer reopen raises the generation and re-sets the + // flag, so its fence has a different owner. This check stops a stale actor (a + // delayed response cleanup, a retry, or an aged reaper snapshot) from clearing + // a newer reopen's live fence. + // + // A caller that reclaims a stranded flag passes `requireStaleSinceBefore`. The + // write-target guard then re-reads the flag timestamp from the fresh row and + // clears the flag only when that timestamp is still older than the cutoff. This + // re-confirms the strand decision against the live row instead of an aged + // snapshot. + async function clearReopenPendingConsumptionUnderLock( + workspaceId: string, + options: { expectedGeneration: number; requireStaleSinceBefore?: Date }, + ): Promise { + return fenceLifecycleGenerationWrite({ + workspaceId, + expectedGeneration: options.expectedGeneration, + isWriteTarget: (fresh) => { + if (!metadataHasReopenPendingConsumption(fresh.metadata)) return false; + if (options.requireStaleSinceBefore) { + const freshSince = readMetadataReopenPendingConsumptionSince(fresh.metadata); + const stillStale = + freshSince === null + || freshSince.getTime() <= options.requireStaleSinceBefore.getTime(); + // The live row shows a fresh flag, so the consumer is still in flight. + if (!stillStale) return false; + } + return true; + }, + onSkip: () => false, + write: async ({ tx, fresh }) => { + await tx + .update(executionWorkspaces) + .set({ + metadata: clearMetadataReopenPendingConsumption(fresh.metadata), + updatedAt: new Date(), + }) + .where(eq(executionWorkspaces.id, workspaceId)); + return true; + }, + }); + } + + // Re-stamp the reopen-pending timestamp for a workspace whose consuming request + // is still in flight. The request that reopens and consumes the worktree is an + // HTTP request, not a heartbeat run, so `workspaceHasActiveRun` cannot see it. + // Without a refresh, a request that runs longer than the stale grace period lets + // the terminal reaper clear the live fence, and a later sweep archives and + // destroys the worktree under the request. The consuming route calls this on an + // interval shorter than the grace period, so the flag never looks stale while the + // request lives. The refresh runs under the lifecycle lock and re-stamps the + // timestamp only while the flag is still set and the current generation still + // equals `expectedGeneration`, so it never revives a cleared flag and never + // refreshes a newer reopen's fence. It returns true when it re-stamped the flag. + async function refreshReopenPendingConsumptionUnderLock( + workspaceId: string, + options: { expectedGeneration: number }, + ): Promise { + return fenceLifecycleGenerationWrite({ + workspaceId, + expectedGeneration: options.expectedGeneration, + // A newer reopen or an archive raised the generation. The gateway then + // skips, so this refresh never re-stamps another owner's fence. + isWriteTarget: (fresh) => metadataHasReopenPendingConsumption(fresh.metadata), + onSkip: () => false, + write: async ({ tx, fresh }) => { + await tx + .update(executionWorkspaces) + .set({ + metadata: setMetadataReopenPendingConsumption(fresh.metadata, now()), + updatedAt: new Date(), + }) + .where(eq(executionWorkspaces.id, workspaceId)); + return true; + }, + }); + } + + async function cleanupTerminalWorkspace( + workspace: ExecutionWorkspaceRow, + expectedHeadSha: string | null, + capturedGeneration: number, + ): Promise<{ cleaned: boolean; warnings: string[]; skippedReopened?: boolean }> { + // The gateway holds the per-workspace lifecycle lock across the destructive + // actions. A reopen takes the same lock, so a reopen cannot rebuild the + // worktree while this cleanup runs, and this cleanup cannot delete a worktree + // that a reopen already restored. The advisory lock (not a row FOR UPDATE) + // gives the exclusion, so the cleanup body can still update the same row on + // the pooled connection without a self-block. A reopen restored this + // workspace after it was archived when the guard fails, so the cleanup skips + // and does not destroy the rebuilt worktree. + return fenceLifecycleGenerationWrite<{ cleaned: boolean; warnings: string[]; skippedReopened?: boolean }>({ + workspaceId: workspace.id, + expectedGeneration: capturedGeneration, + isWriteTarget: (fresh) => isClosedExecutionWorkspaceStatus(fresh.status), + skipLog: { + event: "execution_workspace.cleanup_skipped", + message: "execution workspace cleanup skipped because it was reopened", + }, + onSkip: () => ({ cleaned: false, warnings: [], skippedReopened: true }), + write: () => runTerminalWorkspaceCleanup(workspace, expectedHeadSha), + }); + } + + async function runTerminalWorkspaceCleanup(workspace: ExecutionWorkspaceRow, expectedHeadSha: string | null) { const [ { acquireGitWorktreeCleanupLock, @@ -1367,6 +1666,46 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic } } + // Write the cleanup-failed status under the per-workspace lifecycle lock, but + // only while the row is still closed at the generation the reaper captured. The + // reaper calls this from its catch handler after a cleanup threw. The cleanup + // transaction already rolled back and released the lock, so a reopen can run in + // the window before this write. A reopen raises the generation and restores the + // row to active. A later archive lowers the row back to closed but keeps the + // higher generation. The generation compare then skips this write, so stale + // cleanup-failure state never lands on a newer archive lifecycle. The function + // returns true when it wrote the status, or false when the fence skipped it. + async function markTerminalCleanupFailedFenced(input: { + workspaceId: string; + capturedGeneration: number; + cleanupReason: string; + }): Promise { + // A reopen restored the row after the cleanup threw when the guard fails. The + // gateway then skips, so the stale cleanup-failure status never overwrites the + // newer lifecycle state. + return fenceLifecycleGenerationWrite({ + workspaceId: input.workspaceId, + expectedGeneration: input.capturedGeneration, + isWriteTarget: (fresh) => isClosedExecutionWorkspaceStatus(fresh.status), + skipLog: { + event: "execution_workspace.cleanup_failed_write_skipped", + message: "execution workspace cleanup-failure write skipped because it was reopened", + }, + onSkip: () => false, + write: async ({ tx }) => { + await tx + .update(executionWorkspaces) + .set({ + status: "cleanup_failed", + cleanupReason: input.cleanupReason, + updatedAt: now(), + }) + .where(eq(executionWorkspaces.id, input.workspaceId)); + return true; + }, + }); + } + function buildListConditions( companyId: string, filters?: { @@ -2069,6 +2408,8 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic skippedNonTerminalTree: 0, skippedUndelivered: 0, skippedRace: 0, + skippedReopened: 0, + clearedStaleReopenPending: 0, }; } terminalSweepInProgress = true; @@ -2130,13 +2471,30 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic skippedNonTerminalTree: 0, skippedUndelivered: 0, skippedRace: 0, + skippedReopened: 0, + clearedStaleReopenPending: 0, }; for (const workspace of candidates) { const executionWorkspace = toExecutionWorkspace(workspace); const { git } = await inspectGitCloseReadiness(executionWorkspace); const assessment = await assessDelivery(workspace, git); + const reopenPending = metadataHasReopenPendingConsumption( + workspace.metadata as Record | null, + ); if (!assessment.sourceIssueTerminal || !assessment.subtreeTerminal) { + if (reopenPending) { + // The source issue left the terminal state, so the reopen transition + // committed. Clear the reopen-pending flag under the lifecycle lock so + // a later terminal cycle can archive the workspace again. Pass the + // generation from this snapshot, so the clear skips a newer reopen that + // raised the generation and installed its own fence after this read. + await clearReopenPendingConsumptionUnderLock(workspace.id, { + expectedGeneration: readExecutionWorkspaceLifecycleGeneration( + workspace.metadata as Record | null, + ), + }); + } result.skippedNonTerminalTree += 1; continue; } @@ -2151,64 +2509,135 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic result.skippedUndelivered += 1; continue; } + if (reopenPending) { + const pendingSince = readMetadataReopenPendingConsumptionSince( + workspace.metadata as Record | null, + ); + const staleBefore = new Date(now().getTime() - STALE_REOPEN_PENDING_CONSUMPTION_GRACE_MS); + const stranded = + pendingSince === null + || pendingSince.getTime() <= staleBefore.getTime(); + if (!stranded) { + // A reopen published this workspace as active while the source issue + // is still terminal, and the consuming request is still in flight. Do + // not archive it now. The authoritative check is the NOT reopenPending + // predicate in the archive statement below; this early skip avoids the + // work when the snapshot already shows the flag. + result.skippedReopened += 1; + continue; + } + // The flag is older than the grace period, but age alone does not prove + // the consuming request ended. An authorized request can run longer than + // the grace period. A live consuming run still owns the fence, so keep it + // and skip. This stops the sweep from clearing the fence of a slow request + // that a later sweep would then archive and destroy under the request. + if (await workspaceHasActiveRun(workspace)) { + result.skippedReopened += 1; + continue; + } + // The reopen-pending flag outlived its consumption window and no run owns + // it. The consuming request ended without moving the issue out of the + // terminal state, or the server exited before it cleared the flag. Clear + // the stranded flag under the lifecycle lock so a later sweep can reclaim + // the workspace. Keep the row active, so a retried resume can still reuse + // the rebuilt worktree. Pass this snapshot's generation and re-confirm + // staleness against the fresh row under the lock, so a newer reopen that + // raised the generation or refreshed the timestamp keeps its live fence. + const cleared = await clearReopenPendingConsumptionUnderLock(workspace.id, { + expectedGeneration: readExecutionWorkspaceLifecycleGeneration( + workspace.metadata as Record | null, + ), + requireStaleSinceBefore: staleBefore, + }); + if (cleared) { + result.clearedStaleReopenPending += 1; + logger.info( + { + event: "execution_workspace.reopen", + outcome: "stale_reopen_pending_cleared", + executionWorkspaceId: workspace.id, + sourceIssueId: workspace.sourceIssueId, + pendingSince: pendingSince?.toISOString() ?? null, + }, + "cleared a stranded reopen-pending flag on a terminal workspace", + ); + } + continue; + } if (await workspaceHasActiveRun(workspace)) { result.skippedActiveRun += 1; continue; } result.eligible += 1; const closedAt = now(); - const archived = await db - .update(executionWorkspaces) - .set({ - status: "archived", - closedAt, - cleanupEligibleAt: workspace.cleanupEligibleAt ?? closedAt, - cleanupReason: ISSUE_TERMINAL_WORKSPACE_CLEANUP_REASON, - updatedAt: closedAt, - }) - .where(and( - eq(executionWorkspaces.id, workspace.id), - eq(executionWorkspaces.companyId, workspace.companyId), - inArray(executionWorkspaces.status, ["active", "idle", "in_review"]), - isNull(executionWorkspaces.closedAt), - sql`EXISTS ( - SELECT 1 - FROM ${issues} source_issue - WHERE source_issue.company_id = ${workspace.companyId} - AND source_issue.id = ${workspace.sourceIssueId} - AND source_issue.status IN ('done', 'cancelled') - )`, - sql`NOT EXISTS ( - SELECT 1 - FROM ${issues} linked_issue - JOIN ${heartbeatRuns} live_run - ON live_run.id = linked_issue.checkout_run_id - OR live_run.id = linked_issue.execution_run_id - WHERE linked_issue.company_id = ${workspace.companyId} - AND ( - linked_issue.execution_workspace_id = ${workspace.id} - OR linked_issue.id = ${workspace.sourceIssueId} + // Raise the lifecycle generation on archive. The cleanup below captures + // this generation and re-checks it before it deletes the worktree, so a + // reopen that runs in between fences the cleanup off. + const archivedMetadata = bumpExecutionWorkspaceLifecycleGeneration( + workspace.metadata as Record | null, + ); + // Take the per-workspace lifecycle lock before the archive decision, so a + // concurrent reopen cannot publish an active row between the predicate + // checks and the archive write. The archive statement re-checks the + // status, the terminal predicates, and the reopen-pending flag under the + // lock, so it never archives a workspace that a reopen just restored. + const archived = await db.transaction(async (tx) => { + await acquireExecutionWorkspaceLifecycleLock(tx, workspace.id); + return tx + .update(executionWorkspaces) + .set({ + status: "archived", + closedAt, + cleanupEligibleAt: workspace.cleanupEligibleAt ?? closedAt, + cleanupReason: ISSUE_TERMINAL_WORKSPACE_CLEANUP_REASON, + metadata: archivedMetadata, + updatedAt: closedAt, + }) + .where(and( + eq(executionWorkspaces.id, workspace.id), + eq(executionWorkspaces.companyId, workspace.companyId), + inArray(executionWorkspaces.status, ["active", "idle", "in_review"]), + isNull(executionWorkspaces.closedAt), + sql`(${executionWorkspaces.metadata} ->> ${EXECUTION_WORKSPACE_REOPEN_PENDING_METADATA_KEY}) IS DISTINCT FROM 'true'`, + sql`EXISTS ( + SELECT 1 + FROM ${issues} source_issue + WHERE source_issue.company_id = ${workspace.companyId} + AND source_issue.id = ${workspace.sourceIssueId} + AND source_issue.status IN ('done', 'cancelled') + )`, + sql`NOT EXISTS ( + SELECT 1 + FROM ${issues} linked_issue + JOIN ${heartbeatRuns} live_run + ON live_run.id = linked_issue.checkout_run_id + OR live_run.id = linked_issue.execution_run_id + WHERE linked_issue.company_id = ${workspace.companyId} + AND ( + linked_issue.execution_workspace_id = ${workspace.id} + OR linked_issue.id = ${workspace.sourceIssueId} + ) + AND live_run.company_id = ${workspace.companyId} + AND live_run.status IN ('queued', 'running') + )`, + sql`NOT EXISTS ( + WITH RECURSIVE issue_tree(id, status) AS ( + SELECT root.id, root.status + FROM ${issues} root + WHERE root.company_id = ${workspace.companyId} + AND root.id = ${workspace.sourceIssueId} + UNION ALL + SELECT child.id, child.status + FROM ${issues} child + JOIN issue_tree parent ON child.parent_id = parent.id + WHERE child.company_id = ${workspace.companyId} ) - AND live_run.company_id = ${workspace.companyId} - AND live_run.status IN ('queued', 'running') - )`, - sql`NOT EXISTS ( - WITH RECURSIVE issue_tree(id, status) AS ( - SELECT root.id, root.status - FROM ${issues} root - WHERE root.company_id = ${workspace.companyId} - AND root.id = ${workspace.sourceIssueId} - UNION ALL - SELECT child.id, child.status - FROM ${issues} child - JOIN issue_tree parent ON child.parent_id = parent.id - WHERE child.company_id = ${workspace.companyId} - ) - SELECT 1 FROM issue_tree WHERE status NOT IN ('done', 'cancelled') - )`, - )) - .returning() - .then((rows) => rows[0] ?? null); + SELECT 1 FROM issue_tree WHERE status NOT IN ('done', 'cancelled') + )`, + )) + .returning() + .then((rows) => rows[0] ?? null); + }); if (!archived) { result.skippedRace += 1; continue; @@ -2229,21 +2658,27 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic }, }); + const capturedGeneration = readExecutionWorkspaceLifecycleGeneration( + archived.metadata as Record | null, + ); try { - const cleanup = await cleanupTerminalWorkspace(archived, assessment.workspaceHeadSha); - if (!cleanup.cleaned) result.cleanupFailed += 1; + const cleanup = await cleanupTerminalWorkspace(archived, assessment.workspaceHeadSha, capturedGeneration); + if (cleanup.skippedReopened) result.skippedReopened += 1; + else if (!cleanup.cleaned) result.cleanupFailed += 1; else result.archived += 1; } catch (error) { result.cleanupFailed += 1; const failure = error instanceof Error ? error.message : String(error); - await db - .update(executionWorkspaces) - .set({ - status: "cleanup_failed", - cleanupReason: `${ISSUE_TERMINAL_WORKSPACE_CLEANUP_REASON} | ${failure}`, - updatedAt: now(), - }) - .where(eq(executionWorkspaces.id, archived.id)); + // Mark cleanup_failed only while the row is still closed at the + // generation this sweep captured. A reopen that raced after the failure + // raises the generation and restores the row; a later archive keeps the + // higher generation. The fenced write then skips, so stale + // cleanup-failure state never lands on a newer archive lifecycle. + await markTerminalCleanupFailedFenced({ + workspaceId: archived.id, + capturedGeneration, + cleanupReason: `${ISSUE_TERMINAL_WORKSPACE_CLEANUP_REASON} | ${failure}`, + }); await logActivity(db, { companyId: archived.companyId, actorType: "system", @@ -2280,6 +2715,466 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic return row ? toExecutionWorkspace(row) : null; }, + // Reopen one closed isolated execution workspace so an authorized issue can + // use it again. The caller must first authorize the request on the issue. + // The whole operation runs under the per-workspace lifecycle lock: it reads + // the row in the issue scope, rebuilds the worktree, and only then publishes + // the row as "active". A rebuild failure keeps the row closed and returns an + // error, so the caller never dispatches a run against a broken workspace. + reopenClosedIsolatedExecutionWorkspaceForIssue: async (input: { + workspaceId: string; + issue: { id: string; companyId: string; projectId: string | null }; + actor: { agentId: string | null; actorType: string }; + }): Promise => { + const { issue, actor } = input; + // Bind the workspace to the issue company and project. A null project on + // the issue must match a null project on the row (IS NOT DISTINCT FROM). + const projectIdCondition = + issue.projectId == null + ? sql`${executionWorkspaces.projectId} IS NULL` + : eq(executionWorkspaces.projectId, issue.projectId); + + return db.transaction(async (tx): Promise => { + await acquireExecutionWorkspaceLifecycleLock(tx, input.workspaceId); + const row = await tx + .select() + .from(executionWorkspaces) + .where(and( + eq(executionWorkspaces.id, input.workspaceId), + eq(executionWorkspaces.companyId, issue.companyId), + projectIdCondition, + eq(executionWorkspaces.mode, "isolated_workspace"), + )) + .then((rows) => rows[0] ?? null); + if (!row) { + // Wrong company, wrong project, wrong mode, or missing. Fail closed and + // disclose no workspace detail. + return { ok: false, code: "not_reopenable", message: "Execution workspace is not reopenable" }; + } + if (!isClosedExecutionWorkspaceStatus(row.status)) { + // A concurrent reopen already restored the row. Report success without a + // second rebuild so the caller continues normally. The other request + // owns the reopen-pending flag, so return its generation and let the + // caller skip the consumption guard. + return { + ok: true, + reopened: false, + workspace: toExecutionWorkspace(row), + generation: readExecutionWorkspaceLifecycleGeneration(row.metadata as Record | null), + }; + } + + const [{ ensurePersistedExecutionWorkspaceAvailable }, { workspaceOperationService }] = + await Promise.all([ + import("./workspace-runtime.js"), + import("./workspace-operations.js"), + ]); + const [projectWorkspace, projectPolicy] = await Promise.all([ + row.projectWorkspaceId + ? db + .select({ cwd: projectWorkspaces.cwd }) + .from(projectWorkspaces) + .where(and( + eq(projectWorkspaces.companyId, row.companyId), + eq(projectWorkspaces.id, row.projectWorkspaceId), + )) + .then((rows) => rows[0] ?? null) + : null, + db + .select({ executionWorkspacePolicy: projects.executionWorkspacePolicy }) + .from(projects) + .where(and(eq(projects.companyId, row.companyId), eq(projects.id, row.projectId))) + .then((rows) => parseProjectExecutionWorkspacePolicy(rows[0]?.executionWorkspacePolicy)), + ]); + const config = readExecutionWorkspaceConfig(row.metadata as Record | null); + const nextGeneration = readExecutionWorkspaceLifecycleGeneration( + row.metadata as Record | null, + ) + 1; + const nextMetadata = bumpExecutionWorkspaceLifecycleGeneration( + row.metadata as Record | null, + ); + const recorder = workspaceOperationService(db).createRecorder({ + companyId: row.companyId, + executionWorkspaceId: row.id, + }); + + let rebuildError: string | null = null; + try { + const realized = await ensurePersistedExecutionWorkspaceAvailable({ + db: tx as unknown as Db, + base: { + baseCwd: projectWorkspace?.cwd ?? row.cwd ?? "", + source: "task_session", + projectId: row.projectId, + workspaceId: row.projectWorkspaceId, + repoUrl: row.repoUrl, + repoRef: row.baseRef, + }, + workspace: { + id: row.id, + mode: row.mode, + strategyType: row.strategyType, + cwd: row.cwd, + providerRef: row.providerRef, + projectId: row.projectId, + projectWorkspaceId: row.projectWorkspaceId, + repoUrl: row.repoUrl, + baseRef: row.baseRef, + branchName: row.branchName, + metadata: row.metadata as Record | null, + config: { + ...config, + provisionCommand: + config?.provisionCommand + ?? projectPolicy?.workspaceStrategy?.provisionCommand + ?? null, + }, + }, + issue: row.sourceIssueId + ? { id: row.sourceIssueId, identifier: null, title: row.name } + : null, + agent: { + id: actor.agentId ?? null, + name: actor.actorType === "user" ? "Board" : "Agent", + companyId: row.companyId, + }, + recorder, + }); + if (!realized) { + rebuildError = "Execution workspace could not be rebuilt"; + } + } catch (error) { + rebuildError = error instanceof Error ? error.message : String(error); + } + + if (rebuildError) { + // The rebuild failed. Keep the row closed and retryable. Raise the + // generation so a queued cleanup that captured the old generation does + // nothing. Clear cleanupEligibleAt so the reaper does not destroy the + // half-built worktree while a later reopen retries. + await tx + .update(executionWorkspaces) + .set({ + cleanupReason: EXECUTION_WORKSPACE_REOPEN_FAILED_REASON, + cleanupEligibleAt: null, + metadata: nextMetadata, + updatedAt: new Date(), + }) + .where(eq(executionWorkspaces.id, row.id)); + // The server log carries the underlying cause for diagnosis. The audit + // event and the returned message stay free of repo URLs, host paths, + // and git output. + logger.warn( + { + event: "execution_workspace.reopen", + outcome: "rebuild_failed", + executionWorkspaceId: row.id, + issueId: issue.id, + companyId: row.companyId, + actorType: actor.actorType, + actorAgentId: actor.agentId ?? null, + generation: nextGeneration, + error: rebuildError, + }, + "execution workspace reopen rebuild failed", + ); + await logActivity(tx as unknown as Db, { + companyId: row.companyId, + actorType: actor.actorType === "user" ? "user" : "agent", + actorId: actor.agentId ?? "system", + agentId: actor.actorType === "user" ? null : actor.agentId, + action: "execution_workspace.reopen_failed", + entityType: "execution_workspace", + entityId: row.id, + details: { + issueId: issue.id, + outcome: "rebuild_failed", + generation: nextGeneration, + }, + }); + return { ok: false, code: "rebuild_failed", message: "Failed to rebuild the execution workspace" }; + } + + // The rebuild succeeded. Publish the row as active in one write, and clear + // the closed markers. Set the reopen-pending flag in the same write. The + // source issue is still terminal at this point, because the route changes + // the issue out of the terminal state only after this reopen returns. The + // flag stops the terminal reaper and the archive route from archiving and + // destroying the rebuilt worktree in that window. + const activeMetadata = setMetadataReopenPendingConsumption(nextMetadata, now()); + const activeRow = await tx + .update(executionWorkspaces) + .set({ + status: "active", + closedAt: null, + cleanupReason: null, + cleanupEligibleAt: null, + metadata: activeMetadata, + lastUsedAt: new Date(), + updatedAt: new Date(), + }) + .where(eq(executionWorkspaces.id, row.id)) + .returning() + .then((rows) => rows[0] ?? null); + if (!activeRow) { + return { ok: false, code: "rebuild_failed", message: "Failed to rebuild the execution workspace" }; + } + logger.info( + { + event: "execution_workspace.reopen", + outcome: "reopened", + executionWorkspaceId: row.id, + issueId: issue.id, + companyId: row.companyId, + actorType: actor.actorType, + actorAgentId: actor.agentId ?? null, + generation: nextGeneration, + }, + "execution workspace reopened", + ); + await logActivity(tx as unknown as Db, { + companyId: row.companyId, + actorType: actor.actorType === "user" ? "user" : "agent", + actorId: actor.agentId ?? "system", + agentId: actor.actorType === "user" ? null : actor.agentId, + action: "execution_workspace.reopened", + entityType: "execution_workspace", + entityId: row.id, + details: { + issueId: issue.id, + outcome: "reopened", + generation: nextGeneration, + }, + }); + return { ok: true, reopened: true, workspace: toExecutionWorkspace(activeRow), generation: nextGeneration }; + }); + }, + + // Clear the reopen-pending flag after a caller failed to consume a reopened + // workspace. A reopen publishes the rebuilt worktree as active and sets the + // flag while the source issue is still terminal. The route then moves the + // issue out of the terminal state, and the terminal reaper clears the flag + // once it observes the non-terminal issue. If that move never lands (the + // route mutation returns null, throws, or leaves the issue terminal), the + // issue stays terminal and the flag stays set. The terminal reaper and the + // archive route both skip a reopen-pending row, so the rebuilt worktree leaks + // and no path can reclaim it. This method clears the flag under the lifecycle + // lock, so the reaper can archive and reclaim the worktree. It keeps the row + // active, so a retried resume can still reuse the rebuilt worktree. The + // method is idempotent: it does nothing when the flag is already clear. + // Re-stamp the reopen-pending timestamp while a consuming request is still in + // flight. The consuming route calls this on an interval shorter than the stale + // grace period, so the terminal reaper never treats the live fence as stranded. + // The refresh runs only while the flag is still set and the generation still + // matches `expectedGeneration`, so it never revives a cleared flag and never + // refreshes a newer reopen's fence. It returns { refreshed } so the caller can + // stop the interval once the fence is no longer its own. + refreshReopenPendingConsumption: async (input: { + workspaceId: string; + expectedGeneration: number; + }): Promise<{ refreshed: boolean }> => { + const refreshed = await refreshReopenPendingConsumptionUnderLock(input.workspaceId, { + expectedGeneration: input.expectedGeneration, + }); + return { refreshed }; + }, + + clearReopenPendingConsumptionForUnconsumedReopen: async (input: { + workspaceId: string; + issue: { id: string; companyId: string }; + actor: { agentId: string | null; actorType: string }; + // The generation the reopen published the active row at. It owns the flag. + // The clear runs only while the current generation still matches, so it never + // clears a newer reopen's fence. + expectedGeneration: number; + }): Promise<{ cleared: boolean }> => { + const cleared = await clearReopenPendingConsumptionUnderLock(input.workspaceId, { + expectedGeneration: input.expectedGeneration, + }); + if (cleared) { + logger.info( + { + event: "execution_workspace.reopen", + outcome: "unconsumed_reopen_cleared", + executionWorkspaceId: input.workspaceId, + issueId: input.issue.id, + companyId: input.issue.companyId, + actorType: input.actor.actorType, + actorAgentId: input.actor.agentId ?? null, + }, + "execution workspace reopen-pending flag cleared after an unconsumed reopen", + ); + await logActivity(db, { + companyId: input.issue.companyId, + actorType: input.actor.actorType === "user" ? "user" : "agent", + actorId: input.actor.agentId ?? "system", + agentId: input.actor.actorType === "user" ? null : input.actor.agentId, + action: "execution_workspace.reopen_unconsumed", + entityType: "execution_workspace", + entityId: input.workspaceId, + details: { + issueId: input.issue.id, + outcome: "unconsumed_reopen_cleared", + }, + }); + } + return { cleared }; + }, + + // Archive one workspace under the per-workspace lifecycle lock. The archive + // route calls this so the transition to archived and the destruction fence + // both run under the same lock as a reopen. The lock stops a concurrent + // reopen from publishing an active row between the status re-check and the + // archive write. The archive runs only while the row is still open (not + // already archived by a race) and clears the reopen-pending flag. + // + // The method refuses to archive a row that carries the reopen-pending flag. + // A reopen sets that flag when it publishes a rebuilt worktree as active + // while the source issue is still terminal. The method returns a distinct + // "reopen_pending" outcome for that row. It does not clear the flag and does + // not archive. The route maps that outcome to HTTP 409 and returns before any + // destructive cleanup, so the archive control never removes a rebuilt + // worktree during the reopen consumption window. + archiveWorkspaceUnderLifecycleLock: async (input: { + id: string; + patch: Partial; + closedAt: Date; + }): Promise< + | { outcome: "archived"; workspace: ExecutionWorkspace; capturedGeneration: number } + | { outcome: "reopen_pending" } + | null + > => { + return db.transaction(async (tx) => { + await acquireExecutionWorkspaceLifecycleLock(tx, input.id); + const fresh = await tx + .select() + .from(executionWorkspaces) + .where(eq(executionWorkspaces.id, input.id)) + .then((rows) => rows[0] ?? null); + if (!fresh || isClosedExecutionWorkspaceStatus(fresh.status)) { + // The row is missing or already closed by a concurrent path. Do not + // archive again. + return null; + } + if (metadataHasReopenPendingConsumption(fresh.metadata as Record | null)) { + // A reopen published this row as active while its source issue is still + // terminal. A caller will consume the rebuilt worktree. Refuse the + // archive and keep the flag, so the destructive path never removes the + // rebuilt worktree. The route maps this to HTTP 409. + return { outcome: "reopen_pending" }; + } + const baseMetadata = + (input.patch.metadata as Record | null | undefined) + ?? (fresh.metadata as Record | null); + const archiveMetadata = clearMetadataReopenPendingConsumption( + bumpExecutionWorkspaceLifecycleGeneration(baseMetadata), + ); + const archived = await tx + .update(executionWorkspaces) + .set({ + ...input.patch, + status: "archived", + closedAt: input.closedAt, + cleanupReason: null, + metadata: archiveMetadata, + updatedAt: new Date(), + }) + .where(and( + eq(executionWorkspaces.id, input.id), + // Defense in depth: never archive a reopen-pending row even if the + // flag appears between the read above and this write. The lifecycle + // lock already serializes reopen and archive, so this predicate only + // adds a second, authoritative guard at the write. + sql`(${executionWorkspaces.metadata} ->> ${EXECUTION_WORKSPACE_REOPEN_PENDING_METADATA_KEY}) IS DISTINCT FROM 'true'`, + )) + .returning() + .then((rows) => rows[0] ?? null); + if (!archived) return null; + return { + outcome: "archived", + workspace: toExecutionWorkspace(archived), + capturedGeneration: readExecutionWorkspaceLifecycleGeneration(archiveMetadata), + }; + }); + }, + + // Apply the terminal cleanup outcome to a workspace row through the lifecycle + // gateway. The archive route calls this after the destruction fence ran, to + // record cleanup warnings and, when the destroy failed, the cleanup_failed + // status. A reopen that raced after the fence returned restores the row to an + // open status and raises the generation. The gateway re-reads the fresh row + // under the lock and writes only while the row is still closed at + // `capturedGeneration`. So a stale cleanup patch never overwrites the closedAt, + // the cleanup reason, or the status of a freshly rebuilt worktree. The method + // returns the updated row, or null when the guard skipped the write. + applyClosedWorkspaceCleanupOutcome: async (input: { + id: string; + closedAt: Date; + capturedGeneration: number; + cleanupReason: string | null; + markCleanupFailed: boolean; + }): Promise => { + // A reopen restored the row after the destruction fence returned when the + // guard fails. The gateway then skips, so the write never overwrites the + // newly active lifecycle state. + return fenceLifecycleGenerationWrite({ + workspaceId: input.id, + expectedGeneration: input.capturedGeneration, + isWriteTarget: (fresh) => isClosedExecutionWorkspaceStatus(fresh.status), + onSkip: () => null, + write: async ({ tx }) => { + const row = await tx + .update(executionWorkspaces) + .set({ + closedAt: input.closedAt, + cleanupReason: input.cleanupReason, + ...(input.markCleanupFailed ? { status: "cleanup_failed" as const } : {}), + updatedAt: new Date(), + }) + .where(eq(executionWorkspaces.id, input.id)) + .returning() + .then((rows) => rows[0] ?? null); + return row ? toExecutionWorkspace(row) : null; + }, + }); + }, + + // Read the lifecycle generation of a workspace row. The archive route captures + // this before it destroys, so it can hand the value to the destruction fence. + readLifecycleGeneration: async (id: string): Promise => { + const row = await db + .select({ metadata: executionWorkspaces.metadata }) + .from(executionWorkspaces) + .where(eq(executionWorkspaces.id, id)) + .then((rows) => rows[0] ?? null); + return row ? readExecutionWorkspaceLifecycleGeneration(row.metadata as Record | null) : null; + }, + + // Run a destructive workspace cleanup through the lifecycle gateway. The + // caller passes the generation it captured at archive time. If a reopen raised + // the generation or restored the row to an open status, the gateway skips the + // destroy callback, so a cleanup never deletes a worktree that a reopen + // rebuilt. + fenceClosedWorkspaceDestruction: async (input: { + workspaceId: string; + capturedGeneration: number; + destroy: () => Promise; + }): Promise<{ skippedReopened: true } | { skippedReopened: false; result: T }> => { + return fenceLifecycleGenerationWrite< + { skippedReopened: true } | { skippedReopened: false; result: T } + >({ + workspaceId: input.workspaceId, + expectedGeneration: input.capturedGeneration, + isWriteTarget: (fresh) => isClosedExecutionWorkspaceStatus(fresh.status), + skipLog: { + event: "execution_workspace.cleanup_skipped", + message: "execution workspace cleanup skipped because it was reopened", + }, + onSkip: () => ({ skippedReopened: true as const }), + write: async () => ({ skippedReopened: false as const, result: await input.destroy() }), + }); + }, + reconcileExecutionWorkspaceBranch: async ( id: string, input: { diff --git a/ui/src/pages/IssueDetail.tsx b/ui/src/pages/IssueDetail.tsx index f35c552f60..4a620be8fc 100644 --- a/ui/src/pages/IssueDetail.tsx +++ b/ui/src/pages/IssueDetail.tsx @@ -208,7 +208,6 @@ import { import { Badge } from "@/components/ui/badge"; import { deriveOriginatingActor, - getClosedIsolatedExecutionWorkspaceMessage, isClosedIsolatedExecutionWorkspace, ISSUE_CONTINUATION_SUMMARY_DOCUMENT_KEY, ONBOARDING_FIRST_TASK_ORIGIN_KIND, @@ -1759,12 +1758,16 @@ export function IssueDetail() { }); const resolvedCompanyId = issue?.companyId ?? selectedCompanyId; const externalObjectsState = useIssueExternalObjects(issue?.id ?? null); - const commentComposerDisabledReason = useMemo(() => { - if (!issue?.currentExecutionWorkspace || !isClosedIsolatedExecutionWorkspace(issue.currentExecutionWorkspace)) { - return null; - } - return getClosedIsolatedExecutionWorkspaceMessage(issue.currentExecutionWorkspace); - }, [issue?.currentExecutionWorkspace]); + // A closed isolated workspace no longer blocks the composer. The server reopens + // the workspace when the next comment or resume arrives, so the composer stays + // enabled and a hint tells the user what happens. + const closedIsolatedWorkspaceReopenPending = useMemo( + () => Boolean( + issue?.currentExecutionWorkspace + && isClosedIsolatedExecutionWorkspace(issue.currentExecutionWorkspace), + ), + [issue?.currentExecutionWorkspace], + ); const { data: commentPages, @@ -4333,7 +4336,10 @@ export function IssueDetail() { : "Assign an agent to wake them for triage while the subtree remains paused." ) : null; - const composerHint = pausedComposerHint; + const reopenComposerHint = closedIsolatedWorkspaceReopenPending + ? "This issue's isolated workspace was archived. Your next comment or resume reopens it and rebuilds the worktree." + : null; + const composerHint = pausedComposerHint ?? reopenComposerHint; const queuedCommentReason: "hold" | "active_run" | "other" = activePauseHold ? "hold" : "active_run"; const canApplyTreeControl = Boolean(treeControlPreview) @@ -5244,7 +5250,7 @@ export function IssueDetail() { currentAssigneeValue={actualAssigneeValue} suggestedAssigneeValue={suggestedAssigneeValue} mentions={mentionOptions} - composerDisabledReason={commentComposerDisabledReason} + composerDisabledReason={null} composerHint={composerHint} queuedCommentReason={queuedCommentReason} onVote={handleCommentVote}