diff --git a/server/src/services/automatic-sandbox-continuation.test.ts b/server/src/services/automatic-sandbox-continuation.test.ts index 18bcf750c3..26b3911b9e 100644 --- a/server/src/services/automatic-sandbox-continuation.test.ts +++ b/server/src/services/automatic-sandbox-continuation.test.ts @@ -145,6 +145,15 @@ const support = await getEmbeddedPostgresTestSupport(); await heartbeatService(db).resumeInterruptedSandboxRuns(); expect(await successors(f.run.id)).toHaveLength(1); }); + it("does not revive old work if a later run completes between preparation and scheduling", async () => { + const f = await seed(); + expect(await prepareAutomaticSandboxContinuation(db, f.run)).not.toBeNull(); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "succeeded", + startedAt: new Date(), finishedAt: new Date(), contextSnapshot: { issueId: f.issueId } }); + const result = await heartbeatService(db).scheduleBoundedRetry(f.run.id); + expect(result.outcome).toBe("not_scheduled"); + expect(await successors(f.run.id)).toHaveLength(0); + }); it("finishes delivery after a crash between retiring the hold and scheduling", async () => { const f = await seed(); expect(await prepareAutomaticSandboxContinuation(db, f.run)).not.toBeNull(); diff --git a/server/src/services/automatic-sandbox-continuation.ts b/server/src/services/automatic-sandbox-continuation.ts index ba7dd30486..d1c0e3eb99 100644 --- a/server/src/services/automatic-sandbox-continuation.ts +++ b/server/src/services/automatic-sandbox-continuation.ts @@ -15,6 +15,17 @@ export async function runHasUnconfirmedRemoteExecution(db: Db, companyId: string return leases.some(lease => lease.provider && lease.provider !== "local" && !hasRemoteTerminationReceipt(lease)); } +export async function hasLaterSandboxExecution(db: Db, run: typeof heartbeatRuns.$inferSelect, issueId: string) { + const [successor] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, run.companyId), + sql`(${heartbeatRuns.retryOfRunId} = ${run.id} or + (coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId') = ${issueId}::text + and ${heartbeatRuns.createdAt} > ${run.createdAt.toISOString()} + and (${heartbeatRuns.startedAt} is not null or ${heartbeatRuns.status} in ('queued', 'running', 'scheduled_retry', 'succeeded'))))`, + )).limit(1); + return Boolean(successor); +} + /** Repair execution ownership independently of whether task admission is open. * A fresh conversation can inspect unknown prior effects; a stopped sandbox is * the prerequisite, not a user accepting responsibility for those effects. */ @@ -39,14 +50,7 @@ export async function prepareAutomaticSandboxContinuation(db: Db, source: typeof const recorded = await recordedRunAdapter(tx as unknown as Db, run); if (recorded && !isConversationAdapter(recorded)) return null; // A later execution owns current task work. Never revive an older request. - const [successor] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( - eq(heartbeatRuns.companyId, run.companyId), - sql`(${heartbeatRuns.retryOfRunId} = ${run.id} or - (coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId') = ${issue.id}::text - and ${heartbeatRuns.createdAt} > ${run.createdAt.toISOString()} - and (${heartbeatRuns.startedAt} is not null or ${heartbeatRuns.status} in ('queued', 'running', 'scheduled_retry', 'succeeded'))))`, - )).limit(1); - if (successor) return null; + if (await hasLaterSandboxExecution(tx as unknown as Db, run, issue.id)) return null; const leases = await tx.select().from(environmentLeases).where(and( eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id), )); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 5150f9b79b..1d81a460ea 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,5 +1,5 @@ import { PROCESS_IDENTITY_RECORDED, recordNativeLocalProcessStop } from "./native-local-process-stop.js"; -import { prepareAutomaticSandboxContinuation, runHasUnconfirmedRemoteExecution, SANDBOX_INFRASTRUCTURE_ERRORS } from "./automatic-sandbox-continuation.js"; +import { hasLaterSandboxExecution, prepareAutomaticSandboxContinuation, runHasUnconfirmedRemoteExecution, SANDBOX_INFRASTRUCTURE_ERRORS } from "./automatic-sandbox-continuation.js"; import { LegacyControllerLeaseLostError, legacyControllerBootId, legacyControllerClaim, hasLiveLegacyController, revokeExpiredLegacyController, watchLegacyControllerLease } from "./legacy-controller-lease.js"; import { completeTerminatedRemoteNativeSessionCleanup } from "../vendor/paperclip-runner/index.js"; import { remoteExecutionHasStopped, remoteTerminationReceipt, stoppedRemoteCleanupScopes } from "./remote-execution-termination.js"; @@ -15063,6 +15063,13 @@ export function heartbeatService( reusedExisting: true, }; } + // Preparation and scheduling can be separated by a crash or another + // admission. Recheck latest execution while holding the issue lock. + if (issueId && parseObject(run.resultJson?.automaticSandboxRecovery).state === "provider_terminated" && + await hasLaterSandboxExecution(tx as unknown as Db, run, issueId)) { + return { outcome: "not_scheduled", reason: "A later execution owns the current task", + errorCode: "issue_execution_lock_changed", issueId, details: { runId: run.id } }; + } if (retryReason === INTERACTION_CONTINUATION_INFRA_RETRY_REASON) { if (issueId) { await tx.execute(