diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 7ac3ff61700..0ae90ed415a 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -459,6 +459,13 @@ continues automatically when the workspace is available. Internal scheduling attempts remain in the run log without conversation cancellation markers, cancellation toasts, or manual Retry controls. Users can keep sending instructions. +A remote task workspace stays held by its last writer run until that writer's +save and sandbox release settle. This hold continues after the writer run ends. +A new run that finds such a recently ended writer waits for the workspace. It does +not fail setup. The wait uses the same one-hour holder cutoff as a quiet live +writer, measured from the time the writer ended. After the cutoff, preparation +fails with the unfinished-writer error and leaves the held files unchanged. + An instance admission drain keeps accepted new turns queued and displays **Waiting for maintenance**. Ending the audited drain requests admission for these existing runs immediately. The shared delivery coordinator owns the drain's diff --git a/server/src/__tests__/heartbeat-remote-workspace-busy.test.ts b/server/src/__tests__/heartbeat-remote-workspace-busy.test.ts index 981fc7d261e..05c13b25a85 100644 --- a/server/src/__tests__/heartbeat-remote-workspace-busy.test.ts +++ b/server/src/__tests__/heartbeat-remote-workspace-busy.test.ts @@ -112,4 +112,61 @@ describe.skipIf(!support.supported)("inherited remote workspace heartbeat wait", } } finally { transactionSpy.mockRestore(); } }); + + async function terminalWriterFixture(finishedAt: Date) { + const userId = randomUUID(); + await db.insert(authUsers).values({ id: userId, name: "Workspace owner", email: `${userId}@example.test`, + createdAt: new Date(), updatedAt: new Date() }); + const [company] = await db.insert(companies).values({ name: "Settling writer", issuePrefix: `S${randomUUID().slice(0, 6)}`, + requireBoardApprovalForNewAgents: false, defaultResponsibleUserId: userId }).returning(); + await db.insert(companyMemberships).values({ companyId: company.id, principalType: "user", principalId: userId, + status: "active", membershipRole: "owner" }); + const [environment] = await db.insert(environments).values({ name: `Settling ${randomUUID()}`, + driver: "sandbox", status: "active", config: { provider: "daytona" } }).returning(); + const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Native worker", status: "idle", + adapterType: "paperclip_runner", defaultEnvironmentId: environment.id, + adapterConfig: { provider: "codex", model: "gpt-6.1-sol" }, runtimeConfig: { heartbeat: { maxConcurrentRuns: 2 } } }).returning(); + const [issue] = await db.insert(issues).values({ companyId: company.id, title: "Continue after a failed save", + status: "todo", assigneeAgentId: agent.id, responsibleUserId: userId }).returning(); + // The previous run on this task ended, but its workspace save and sandbox + // release have not settled: the writer hold and unsaved files remain. + const [writer] = await db.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent.id, + nativeIssueId: issue.id, runtimeMode: "native", status: "failed", errorCode: "provider_transport_failed", + startedAt: new Date(finishedAt.getTime() - 60_000), lastOutputAt: new Date(finishedAt.getTime() - 30_000), + finishedAt }).returning(); + const authority = { schema: "paperclip.remote-workspace.v1" as const, epoch: 5, snapshotId: randomUUID(), + source: { kind: "empty" as const }, location: null, writerRunId: writer.id, dirty: true }; + const [workspace] = await db.insert(executionWorkspaces).values({ companyId: company.id, sourceIssueId: issue.id, + name: "Task workspace", mode: "isolated_workspace", strategyType: "project_primary", providerType: "cloud_sandbox", + cwd: `/workspaces/${randomUUID()}`, metadata: { remoteWorkspace: authority } }).returning(); + await db.update(issues).set({ executionWorkspaceId: workspace.id, executionWorkspacePreference: "reuse_existing", + executionWorkspaceSettings: { mode: "isolated_workspace" } }).where(eq(issues.id, issue.id)); + const run = await heartbeat.wakeup(agent.id, { source: "on_demand", triggerDetail: "manual", reason: "manual", + requestedByActorType: "user", requestedByActorId: userId, payload: { issueId: issue.id }, contextSnapshot: { issueId: issue.id } }); + expect(run).not.toBeNull(); + await heartbeat.drainActiveRunExecutions(); + const [settled] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run!.id)); + const [current] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspace.id)); + return { run: settled, writer, workspace, authority, current }; + } + + it("waits for a writer that just ended while its save or sandbox release settles", async () => { + const f = await terminalWriterFixture(new Date()); + expect(f.run).toMatchObject({ status: "cancelled", errorCode: "workspace_busy", + error: expect.stringContaining("ended, but its workspace save has not settled"), + resultJson: { executionRecovery: { kind: "workspace_wait", providerWorkStarted: false }, + workspaceBusy: { executionWorkspaceId: f.workspace.id, holderRunId: f.writer.id } } }); + const retries = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, f.run.id)); + expect(retries).toHaveLength(1); + expect(retries[0]).toMatchObject({ status: "scheduled_retry", scheduledRetryReason: "workspace_busy" }); + // Waiting never takes over or discards the unsaved writer's files. + expect(readRemoteWorkspaceAuthority(f.current.metadata)).toEqual(f.authority); + }); + + it("stops waiting for an ended writer whose save stayed unsettled past the holder cutoff", async () => { + const f = await terminalWriterFixture(new Date(Date.now() - 2 * 60 * 60_000)); + expect(f.run).toMatchObject({ status: "failed", error: "The workspace still has an unfinished writer or save operation." }); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, f.run.id))).toEqual([]); + expect(readRemoteWorkspaceAuthority(f.current.metadata)).toEqual(f.authority); + }); }); diff --git a/server/src/services/heartbeat/retries.ts b/server/src/services/heartbeat/retries.ts index 011f87b588c..56f44a0ae13 100644 --- a/server/src/services/heartbeat/retries.ts +++ b/server/src/services/heartbeat/retries.ts @@ -204,11 +204,12 @@ export class WorkspaceBusyDeferral extends Error { executionWorkspaceId?: string | null; deferralAttempt: number; wasIssueAssignee: boolean; + holderEnded?: boolean; }) { super( `Shared ${input.executionWorkspaceId ? "execution" : "project"} workspace is busy: run ${input.holder.runId} (issue ${ input.holder.issueIdentifier ?? input.holder.issueId - }) is still running`, + }) ${input.holderEnded ? "ended, but its workspace save has not settled" : "is still running"}`, ); this.name = "WorkspaceBusyDeferral"; this.holder = input.holder; diff --git a/server/src/services/heartbeat/workspace-preparation.test.ts b/server/src/services/heartbeat/workspace-preparation.test.ts index cf32c0b3967..16fc8fd685d 100644 --- a/server/src/services/heartbeat/workspace-preparation.test.ts +++ b/server/src/services/heartbeat/workspace-preparation.test.ts @@ -386,7 +386,15 @@ describe.skipIf(!support.supported && !externalDatabaseUrl)("heartbeat workspace await db.update(heartbeatRuns).set({ lastOutputAt: new Date(Date.now() - WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS - 1_000) }) .where(eq(heartbeatRuns.id, writer.id)); await expect(prepareHeartbeatWorkspace(db, input)).rejects.toMatchObject({ status: 409 }); + // An ended writer whose save has not settled still holds the workspace: + // the next run waits from the writer's end, within the same holder cutoff. await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, writer.id)); + await expect(prepareHeartbeatWorkspace(db, input)).rejects.toMatchObject({ + constructor: WorkspaceBusyDeferral, code: "workspace_busy", executionWorkspaceId: workspace.id, + holder: { runId: writer.id, issueId: parent.id, ended: true }, + }); + await db.update(heartbeatRuns).set({ finishedAt: new Date(Date.now() - WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS - 1_000) }) + .where(eq(heartbeatRuns.id, writer.id)); await expect(prepareHeartbeatWorkspace(db, input)).rejects.toMatchObject({ status: 409 }); expect(await readAuthority()).toEqual(authority); expect(acquire).not.toHaveBeenCalled(); diff --git a/server/src/services/heartbeat/workspace-preparation.ts b/server/src/services/heartbeat/workspace-preparation.ts index 15958e0c6b9..239a05ebceb 100644 --- a/server/src/services/heartbeat/workspace-preparation.ts +++ b/server/src/services/heartbeat/workspace-preparation.ts @@ -1020,7 +1020,7 @@ async function prepareHeartbeatWorkspaceLocked(db: Db, input: HeartbeatWorkspace || error.writer.lastActivityAt.getTime() < Date.now() - WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS) throw error; throw new WorkspaceBusyDeferral({ holder: error.writer, executionWorkspaceId: error.workspaceId, deferralAttempt: run.scheduledRetryReason === WORKSPACE_BUSY_RETRY_REASON ? run.scheduledRetryAttempt ?? 0 : 0, - wasIssueAssignee: input.task.wasIssueAssignee }); + wasIssueAssignee: input.task.wasIssueAssignee, holderEnded: error.writer.ended === true }); }); persistedExecutionWorkspace = await executionWorkspacesSvc.getStoredById(agent.companyId, reserved.id); if (!persistedExecutionWorkspace?.cwd) throw new Error("remote_workspace_reservation_missing"); diff --git a/server/src/services/remote-workspaces.ts b/server/src/services/remote-workspaces.ts index c7c4a00f535..c8da450fbf2 100644 --- a/server/src/services/remote-workspaces.ts +++ b/server/src/services/remote-workspaces.ts @@ -154,6 +154,8 @@ export class RemoteWorkspaceWriterActiveError extends HttpError { readonly workspaceId: string, readonly writer: { runId: string; agentId: string; issueId: string; issueIdentifier: string | null; lastActivityAt: Date; + /** The writer run ended; its workspace save or sandbox release has not settled. */ + ended?: boolean; }, ) { super(409, "The workspace still has an unfinished writer or save operation.", { code: "remote_workspace_writer_pending" }); @@ -219,17 +221,16 @@ export async function reserveRemoteWorkspace(input: { errorCode: heartbeatRuns.errorCode, executionRecovery: sql`${heartbeatRuns.resultJson}->'executionRecovery'`, contextIssueId: sql`coalesce(${heartbeatRuns.contextSnapshot}->>'issueId', ${heartbeatRuns.contextSnapshot}->>'taskId')`, lastOutputAt: heartbeatRuns.lastOutputAt, startedAt: heartbeatRuns.startedAt, createdAt: heartbeatRuns.createdAt, + finishedAt: heartbeatRuns.finishedAt, }).from(heartbeatRuns).where(and( eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, authority.writerRunId))); - if (prior?.status === "running") { - const writerIssueId = prior.nativeIssueId ?? prior.issueId ?? prior.contextIssueId; - const [writerIssue] = writerIssueId ? await tx.select({ id: issues.id, identifier: issues.identifier }).from(issues) - .where(and(eq(issues.companyId, input.companyId), eq(issues.id, canonicalExecutionUuid(sql`${writerIssueId}::text`)))) : []; - if (writerIssue) throw new RemoteWorkspaceWriterActiveError(existing.id, { - runId: authority.writerRunId, agentId: prior.agentId, issueId: writerIssue.id, issueIdentifier: writerIssue.identifier, - lastActivityAt: prior.lastOutputAt ?? prior.startedAt ?? prior.createdAt, - }); - } + const writerIssueId = prior ? prior.nativeIssueId ?? prior.issueId ?? prior.contextIssueId : null; + const [writerIssue] = writerIssueId ? await tx.select({ id: issues.id, identifier: issues.identifier }).from(issues) + .where(and(eq(issues.companyId, input.companyId), eq(issues.id, canonicalExecutionUuid(sql`${writerIssueId}::text`)))) : []; + if (prior?.status === "running" && writerIssue) throw new RemoteWorkspaceWriterActiveError(existing.id, { + runId: authority.writerRunId, agentId: prior.agentId, issueId: writerIssue.id, issueIdentifier: writerIssue.identifier, + lastActivityAt: prior.lastOutputAt ?? prior.startedAt ?? prior.createdAt, + }); const leases = await tx.select().from(environmentLeases).where(and(eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.heartbeatRunId, authority.writerRunId))); const bootstrap = object(prior?.executionRecovery); @@ -240,7 +241,16 @@ export async function reserveRemoteWorkspace(input: { const unusedReservation = !authority.dirty && prior && ["failed", "cancelled", "interrupted", "timed_out", "succeeded"].includes(prior.status) && leases.every(lease => hasRemoteTerminationReceipt(lease) || (pristinePreparation && lease.releasedAt && lease.cleanupStatus === "success" && ["released", "expired", "failed"].includes(lease.status))); - if (!unusedReservation) throw conflict("The workspace still has an unfinished writer or save operation.", { code: "remote_workspace_writer_pending" }); + if (!unusedReservation) { + // An ended writer still holds the workspace until its save and sandbox + // release settle. Report it as the holder so preparation waits within + // the same bounded cutoff as a live writer instead of failing setup. + if (prior && writerIssue) throw new RemoteWorkspaceWriterActiveError(existing.id, { + runId: authority.writerRunId, agentId: prior.agentId, issueId: writerIssue.id, issueIdentifier: writerIssue.identifier, + lastActivityAt: prior.finishedAt ?? prior.lastOutputAt ?? prior.startedAt ?? prior.createdAt, ended: true, + }); + throw conflict("The workspace still has an unfinished writer or save operation.", { code: "remote_workspace_writer_pending" }); + } } if (authority.writerRunId === input.runId) return existing; const next: RemoteWorkspaceAuthority = { ...authority, source, epoch: authority.epoch + 1, writerRunId: input.runId };