diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 34794de52d..1974a4ec33 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -987,3 +987,25 @@ For a board operator, the intended meaning is: - blockers explain waiting That is the execution contract Paperclip should present to operators. + +### Cancellation during native startup + +Cancellation records a preparation fence while holding the run row lock. Native +runtime selection checks that fence, the running status, and the current startup +controller lease in the same transaction that creates the native coordinator. +The native executor rechecks cancellation and terminal status when claiming the +coordinator, before starting or attaching a provider. + +A cancelled startup can continue from a newer authenticated user message after +cleanup. The server requires either its explicit before-selection fence or an +unclaimed native coordinator (zero attempts and controller generations, no +controller, lease, or result). It also checks for contradictory launch/process +evidence and verifies local cleanup or exact remote termination receipts. The +preparer must have finished or its startup lease must have expired. A missing +PID alone does not establish this proof. + +The existing bounded saved-message worker rechecks this proof after restart. +Admission atomically settles an unclaimed coordinator and admits one fresh turn, +preserving history, unknown action outcomes, and attempt counts. Pauses, approvals, +budgets, task ownership, and terminal task status still gate admission. No +automatic provider replay is authorized by a cancelled startup. diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index f94ed46c37..dd6d03fe82 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -2537,6 +2537,69 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { } } + it("fences native selection when cancellation wins during preparation", async () => { + await withTempPaperclipHome(async () => { + const { agentId, issueId, runId } = await seedQueuedIssueRunFixture(); + await db.update(agents).set({ adapterType: "paperclip_runner", + adapterConfig: { provider: "codex", model: "gpt-5.6-luna" }, + }).where(eq(agents.id, agentId)); + const factory = vi.fn(() => { throw new Error("provider must not start"); }); + let reachedSelection = false; + const heartbeat = heartbeatService(db, { + nativeSessionBackendFactory: factory, + beforeNativeRuntimeSelection: async id => { + reachedSelection = true; + await heartbeat.cancelRun(id); + }, + }); + await heartbeat.resumeQueuedRuns(); + await heartbeat.drainActiveRunExecutions(); + expect(reachedSelection).toBe(true); + expect(await heartbeat.getRun(runId)).toMatchObject({ status: "cancelled", runtimeMode: "legacy", + runtimeModeResolvedAt: null, nativeSessionId: null, + resultJson: { startupCancellation: { beforeNativeSelection: true }, + startupPreparationSettledAt: expect.any(String) }, + }); + expect(await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId))).toHaveLength(0); + expect(factory).not.toHaveBeenCalled(); + expect(mockAdapterExecute).not.toHaveBeenCalled(); + const [task] = await db.select().from(issues).where(eq(issues.id, issueId)); + expect(task.executionRunId).toBeNull(); + }); + }); + + it("does not dispatch when cancellation wins after native selection", async () => { + await withTempPaperclipHome(async () => { + const { agentId, issueId, runId } = await seedQueuedIssueRunFixture(); + await db.update(agents).set({ adapterType: "paperclip_runner", + adapterConfig: { provider: "codex", model: "gpt-5.6-luna" }, + }).where(eq(agents.id, agentId)); + await db.update(heartbeatRuns).set({ invocationSource: "automation" }).where(eq(heartbeatRuns.id, runId)); + const factory = vi.fn(() => { throw new Error("provider must not start"); }); + let reachedDispatch = false; + const heartbeat = heartbeatService(db, { + nativeSessionBackendFactory: factory, + beforeChatControlRecoveryCheck: async ({ stage, runId: id }) => { + if (stage !== "dispatch") return; + reachedDispatch = true; + expect((await heartbeat.getRun(id))?.runtimeMode).toBe("native"); + await heartbeat.cancelRun(id); + }, + }); + await heartbeat.resumeQueuedRuns(); + await heartbeat.drainActiveRunExecutions(); + expect(reachedDispatch).toBe(true); + expect(await heartbeat.getRun(runId)).toMatchObject({ status: "cancelled", runtimeMode: "native", + resultJson: { startupPreparationSettledAt: expect.any(String) }, + }); + expect(factory).not.toHaveBeenCalled(); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)); + expect(coordinator).toMatchObject({ attempt: 0, leaseOwner: null }); + const [task] = await db.select().from(issues).where(eq(issues.id, issueId)); + expect(task.executionRunId).toBeNull(); + }); + }); + it("dispatches local native external chat inside the server-selected task root", async () => { await withTempPaperclipHome(async () => { const { companyId, agentId, issueId, runId } = diff --git a/server/src/services/cancelled-native-startup.ts b/server/src/services/cancelled-native-startup.ts new file mode 100644 index 0000000000..6be8792dda --- /dev/null +++ b/server/src/services/cancelled-native-startup.ts @@ -0,0 +1,46 @@ +import { and, eq, inArray, isNotNull, or } from "drizzle-orm"; +import { environmentLeases, heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { claimedAdapterType } from "./conversation-continuation.js"; +import { PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED } from "./native-local-process-stop.js"; +import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; + +type Run = typeof heartbeatRuns.$inferSelect; +type Coordinator = typeof nativeRunFinalizations.$inferSelect; + +/** Caller holds the coordinator and run locks when using this proof to admit + * work. Attempt zero is a durable never-claimed receipt: every native executor + * commits its first claim before it can start or attach a provider. */ +export async function isCancelledNativeStartup(db: Db, run: Run, coordinator: Coordinator | undefined) { + if (run.status !== "cancelled" || !run.finishedAt || run.processPid || run.processGroupId || + run.processStartedAt || run.sessionIdAfter) return false; + const cancellation = run.resultJson?.startupCancellation as Record | undefined; + const beforeSelection = run.runtimeMode === "legacy" && !run.runtimeModeResolvedAt && + !run.nativeSessionId && !coordinator && claimedAdapterType(run) === "paperclip_runner" && + cancellation?.beforeNativeSelection === true; + const neverClaimed = run.runtimeMode === "native" && coordinator && + ["observed", "terminal_failure"].includes(coordinator.phase) && coordinator.attempt === 0 && + coordinator.controllerGeneration === 0 && !coordinator.controllerBootId && + !coordinator.controllerPid && !coordinator.leaseOwner && !coordinator.leaseExpiresAt && + !coordinator.resultId && !coordinator.failureDetail?.successorRunId; + if (!beforeSelection && !neverClaimed) return false; + const settled = typeof run.resultJson?.startupPreparationSettledAt === "string"; + // The old preparer can still be unwinding even though the run is terminal. + if (!settled && run.controllerLeaseExpiresAt && run.controllerLeaseExpiresAt > new Date()) return false; + const leases = await db.select().from(environmentLeases).where(and( + eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id), + )); + if ((!settled && leases.length === 0) || leases.some(lease => + lease.provider === "local" + ? !lease.releasedAt || lease.status === "pending_cleanup" || lease.cleanupStatus === "failed" + : !hasRemoteTerminationReceipt(lease))) return false; + // Reject contradictory retained evidence, including a crash after a launch + // request but before the PID callback. Provider events never certify a stop. + const [execution] = await db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), + or(isNotNull(heartbeatRunEvents.sourceEventId), + inArray(heartbeatRunEvents.eventType, [PROCESS_START_REQUESTED, PROCESS_IDENTITY_RECORDED, + "harness.ready", "session.started", "session.resumed", "session.updated", "turn.started", + "provider.event", "provider.rpc_result", "tool.execution.started"])), + )).limit(1); + return !execution; +} diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts index 607b02c7b3..2f43896571 100644 --- a/server/src/services/explicit-native-continuation.test.ts +++ b/server/src/services/explicit-native-continuation.test.ts @@ -49,6 +49,72 @@ const support = await getEmbeddedPostgresTestSupport(); agentId: f.agentId, status: "queued", contextSnapshot: { issueId: f.issueId, previousRunId: result.previousRunId, forceFreshSession: true } }); return result; }); + async function seedCancelledStartup() { + const f = await seed(); + await db.update(heartbeatRuns).set({ status: "cancelled", processPid: null, + startedAt: new Date("2026-09-11T09:59:59Z"), + runtimeModeResolvedAt: new Date("2026-09-11T10:00:01Z"), + controllerBootId: randomUUID(), controllerLeaseExpiresAt: new Date("2026-09-11T10:01:00Z"), + }).where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.update(nativeRunFinalizations).set({ phase: "observed", attempt: 0, + failureDetail: null, + }).where(eq(nativeRunFinalizations.runId, f.sourceRunId)); + await db.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: f.sourceRunId, + provider: "local", status: "released", releasedAt: new Date("2026-09-11T10:00:02Z"), + cleanupStatus: "succeeded", leasePolicy: "ephemeral" }); + return f; + } + + it("settles a cancelled unclaimed coordinator after restart and admits one user successor", async () => { + const f = await seedCancelledStartup(); + expect(await admit(f, true)).toMatchObject({ previousRunId: f.sourceRunId }); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId)))[0].phase).toBe("observed"); + const results = await Promise.all([admit(f), admit(f)]); + expect(results.filter(Boolean)).toHaveLength(1); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId)); + expect(coordinator).toMatchObject({ phase: "terminal_failure", attempt: 0, + failureCode: "native_startup_cancelled", failureDetail: { replacementDenied: "explicit_user_continuation" } }); + const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + expect(action.evidence.automaticRecovery).toMatchObject({ actionOutcome: "unknown", replay: "explicit_user_continuation" }); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId)))[0].status).toBe("cancelled"); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.successorRunId))).toHaveLength(1); + }); + + it("continues native-runner preparation cancelled before runtime selection", async () => { + const f = await seedCancelledStartup(); + await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId)); + await db.update(heartbeatRuns).set({ runtimeMode: "legacy", runtimeModeResolvedAt: null, nativeIssueId: null, + runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } }, + resultJson: { startupCancellation: { beforeNativeSelection: true }, startupPreparationSettledAt: new Date().toISOString() }, + }).where(eq(heartbeatRuns.id, f.sourceRunId)); + expect(await admit(f)).toMatchObject({ previousRunId: f.sourceRunId }); + }); + + it.each(["attempt", "generation", "controller", "lease", "process", "launch", "provider", "cleanup", "remote", "preparing", "closed", "reassigned"])( + "retains cancellation safeguards with %s evidence", async kind => { + const f = await seedCancelledStartup(); + if (kind === "attempt") await db.update(nativeRunFinalizations).set({ attempt: 1 }).where(eq(nativeRunFinalizations.runId, f.sourceRunId)); + if (kind === "generation") await db.update(nativeRunFinalizations).set({ controllerGeneration: 1 }).where(eq(nativeRunFinalizations.runId, f.sourceRunId)); + if (kind === "controller") await db.update(nativeRunFinalizations).set({ controllerBootId: "old-owner" }).where(eq(nativeRunFinalizations.runId, f.sourceRunId)); + if (kind === "lease") await db.update(nativeRunFinalizations).set({ leaseOwner: "owner", leaseExpiresAt: new Date(Date.now() + 60000) }).where(eq(nativeRunFinalizations.runId, f.sourceRunId)); + if (kind === "process") await db.update(heartbeatRuns).set({ processPid: process.pid }).where(eq(heartbeatRuns.id, f.sourceRunId)); + if (kind === "launch" || kind === "provider") await db.insert(heartbeatRunEvents).values({ companyId: f.companyId, + agentId: f.agentId, runId: f.sourceRunId, seq: 1, + eventType: kind === "launch" ? PROCESS_START_REQUESTED : "provider.event", + ...(kind === "provider" ? { sourceEventId: "provider-1", sourceInstanceId: "provider", sourceSeq: 1, protocolSchemaVersion: 1, canonicalPayloadHash: "hash" } : {}), + }); + if (kind === "cleanup") await db.update(environmentLeases).set({ status: "pending_cleanup", cleanupStatus: "failed" }).where(eq(environmentLeases.heartbeatRunId, f.sourceRunId)); + if (kind === "remote") await db.update(environmentLeases).set({ provider: "daytona", providerLeaseId: "unverified" }).where(eq(environmentLeases.heartbeatRunId, f.sourceRunId)); + if (kind === "preparing") await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: new Date(Date.now() + 60000) }).where(eq(heartbeatRuns.id, f.sourceRunId)); + if (kind === "closed") await db.update(issues).set({ status: "done" }).where(eq(issues.id, f.issueId)); + if (kind === "reassigned") await db.update(issues).set({ assigneeAgentId: null }).where(eq(issues.id, f.issueId)); + expect(await admit(f)).toBeNull(); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId)))[0].phase).toBe("observed"); + await db.delete(environmentLeases).where(eq(environmentLeases.heartbeatRunId, f.sourceRunId)); + }, + ); + it("preserves local stop proof after process metadata is cleared and invalidates it on another launch", async () => { const f = await seed(); const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId)); @@ -85,8 +151,8 @@ const support = await getEmbeddedPostgresTestSupport(); expect(await hasNativeLocalProcessStop(db, f.companyId, source.id)).toBe(false); }); - it("resumes saved local messages after restart exactly once and keeps the same wait receipt while blocked", async () => { - const f = await seed(); + it.each(["stopped_process", "cancelled_startup"])("resumes saved local messages after restart exactly once: %s", async kind => { + const f = kind === "cancelled_startup" ? await seedCancelledStartup() : await seed(); await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }); await db.update(heartbeatRuns).set({ processPid: process.pid }).where(eq(heartbeatRuns.id, f.sourceRunId)); // A prior cancelled admission is also held, but cannot select the native @@ -105,7 +171,7 @@ const support = await getEmbeddedPostgresTestSupport(); requestedByActorType: "user", requestedByActorId: "board", payload: { issueId: f.issueId, commentId: f.commentId }, contextSnapshot: { issueId: f.issueId, wakeCommentId: f.commentId } }); const [waiting] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId)); - expect(waiting.payload?.executionWait).toMatchObject({ reason: "process_running" }); + expect(waiting.payload?.executionWait).toMatchObject({ reason: kind === "cancelled_startup" ? "controller_settling" : "process_running" }); const makeDue = () => db.update(agentWakeupRequests).set({ updatedAt: new Date(0) }).where(eq(agentWakeupRequests.id, waiting.id)); await makeDue(); await heartbeatService(db).resumeExecutionWaitComments(); diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index 6405b9bbc3..3aeaef4b88 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -1,3 +1,4 @@ +import { isCancelledNativeStartup } from "./cancelled-native-startup.js"; import { hasNativeLocalProcessStop } from "./native-local-process-stop.js"; import { completeTerminatedRemoteNativeSessionCleanup } from "../vendor/paperclip-runner/index.js"; import { hasRemoteTerminationReceipt, remoteLeaseCleanupScope } from "./remote-execution-termination.js"; @@ -74,11 +75,12 @@ export async function admitExplicitNativeContinuation(input: { if (pendingInteraction || pendingApproval) return blocked("decision_pending", "A pending approval or question must be resolved before this message can start."); const sources: Run[] = []; + const cancelledStartupIds = new Set(); for (const action of actions) { const runId = action.evidence.runId ?? action.evidence.sourceRunId; if (typeof runId !== "string") return blocked("source_missing", "The stopped run could not be identified. Your message is saved."); // Text comparison keeps malformed historical evidence a hold, not a UUID cast error. - const [run] = await db.select().from(heartbeatRuns).where(and( + let [run] = await db.select().from(heartbeatRuns).where(and( eq(heartbeatRuns.companyId, companyId), sql`${heartbeatRuns.id}::text = ${runId}`, )); if (!run || run.agentId !== agentId || !terminal.includes(run.status) || @@ -101,12 +103,25 @@ export async function admitExplicitNativeContinuation(input: { // For pre-upgrade rows without adapter evidence, only a new explicit user // turn is allowed, after the termination proofs below. This does not infer // an old adapter type, certify old outcomes, or authorize automatic replay. - if (run.runtimeMode !== "native" && !unusedAdmission && !legacyUserTurn) return null; const [coordinator] = await db.select().from(nativeRunFinalizations).where(and( eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, run.id), )).for("update"); - if (coordinator && (coordinator.phase !== "terminal_failure" || coordinator.leaseOwner || - coordinator.resultId || coordinator.failureDetail?.successorRunId)) return blocked("controller_settling", "Waiting for the previous run to finish recovery. Your message will start automatically."); + // Same lock order as the native claim. Re-read the run while holding both + // locks before accepting the never-claimed startup proof. + const [lockedRun] = await db.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, run.id), + )).for("update"); + if (!lockedRun || lockedRun.status !== run.status || lockedRun.agentId !== run.agentId || + lockedRun.finishedAt?.getTime() !== run.finishedAt.getTime()) return null; + run = lockedRun; + const cancelledStartup = await isCancelledNativeStartup(db, run, coordinator); + if (cancelledStartup) cancelledStartupIds.add(run.id); + if (run.runtimeMode !== "native" && !unusedAdmission && !legacyUserTurn && !cancelledStartup) return null; + if (!cancelledStartup && coordinator && (coordinator.phase !== "terminal_failure" || coordinator.leaseOwner || + coordinator.resultId || coordinator.failureDetail?.successorRunId)) return blocked("controller_settling", + run.status === "cancelled" && !coordinator.leaseOwner + ? "The cancelled run still needs verified cleanup. Your message is saved. Inspect the run and its environment for details." + : "Waiting for the previous run to finish recovery. Your message will start automatically."); const leases = await db.select() .from(environmentLeases).where(and( eq(environmentLeases.companyId, companyId), eq(environmentLeases.heartbeatRunId, run.id), @@ -120,7 +135,7 @@ export async function admitExplicitNativeContinuation(input: { }))) return null; } else { if (leases.some(lease => !lease.releasedAt || lease.cleanupStatus === "failed")) return blocked("local_cleanup", "Waiting for the previous environment to finish cleanup. Your message will start automatically."); - if (!unusedAdmission) { + if (!unusedAdmission && !cancelledStartup) { // A missing process identity is not evidence that a provider exited. if (!run.processPid && !run.processGroupId && !await hasNativeLocalProcessStop(db, companyId, run.id)) return blocked("process_identity_missing", "The previous run has no verified stop record. Paperclip cannot start this message yet."); @@ -148,6 +163,16 @@ export async function admitExplicitNativeContinuation(input: { if (input.dryRun) return { previousRunId: previous.id, commentId, ...(retry ? { failedRunId: input.failedRunId! } : {}) }; const authorization = { actorId, commentId, ...(retry ? { failedRunId: input.failedRunId } : {}), runId: input.successorRunId, previousRunId: previous.id, recordedAt: new Date().toISOString() }; + for (const runId of cancelledStartupIds) { + await db.update(nativeRunFinalizations).set({ + phase: "terminal_failure", failureCode: "native_startup_cancelled", nextAttemptAt: null, + controlDeadlineAt: null, updatedAt: new Date(), + }).where(and(eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, runId))); + await db.update(heartbeatRuns).set({ + ...(nativeSources.some(run => run.id === runId) ? { nativePhase: "terminal_failure", nativePhaseUpdatedAt: new Date() } : {}), + executionControlDeadlineAt: null, + }).where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, runId))); + } if (nativeSources.length) await db.update(nativeRunFinalizations).set({ failureDetail: sql`coalesce(${nativeRunFinalizations.failureDetail}, '{}'::jsonb) || ${JSON.stringify({ replacementDenied: "explicit_user_continuation" })}::jsonb`, updatedAt: new Date(), diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 681bdde7c0..09906d335c 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,6 +1,6 @@ import { AGENT_CHAT_DIRECTIVE, conversationReplay, isConversation, isConversationExecutionWake, isWaitingConversation, prepareConversationTurn, settleConversationTurn } from "./agent-conversations.js"; import { PROCESS_IDENTITY_RECORDED, recordNativeLocalProcessStop } from "./native-local-process-stop.js"; -import { legacyControllerBootId, legacyControllerClaim, hasLiveLegacyController, revokeExpiredLegacyController, watchLegacyControllerLease } from "./legacy-controller-lease.js"; +import { legacyControllerBootId, legacyControllerClaim, renewLegacyControllerLease, 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"; import { applyConnectorSkills, prepareConnectorSkillDelivery, resolveConnectorAssignments } from "./connector-runtime.js"; @@ -9044,6 +9044,8 @@ export type HeartbeatEnvironmentRuntime = ReturnType< >; export interface HeartbeatServiceOptions { + /** Test seam before the atomic native runtime handoff. */ + beforeNativeRuntimeSelection?: (runId: string) => Promise; /** Test seam immediately before the durable chat-control admission check. */ beforeChatControlRecoveryCheck?: (input: { runId: string; @@ -10021,7 +10023,9 @@ export function heartbeatService( async function resumeRemoteStopComments(run: typeof heartbeatRuns.$inferSelect, requestId?: string) { if (!isHeartbeatRunTerminalStatus(run.status) || adapterExecutionControls.has(run.id)) return; - if (run.runtimeMode !== "native" && !(await remoteExecutionHasStopped(db, run.companyId, run.id))) return; + if (run.runtimeMode !== "native" && + parseObject(run.resultJson?.startupCancellation).beforeNativeSelection !== true && + !(await remoteExecutionHasStopped(db, run.companyId, run.id))) return; const issueId = run.nativeIssueId ?? (typeof run.contextSnapshot?.issueId === "string" ? run.contextSnapshot.issueId : null); if (!issueId) return; const legacyContinuation = run.runtimeMode === "legacy" && @@ -19367,6 +19371,7 @@ export function heartbeatService( Parameters[0] | null = null; let nativeSessionResumeScheduled = false; let nativeOwnershipHeld = false; + let nativeDispatchStarted = false; let nativeWorkspaceFinalizeScheduled = false; let nativeWorkspaceSync: Awaited< ReturnType @@ -22654,7 +22659,8 @@ export function heartbeatService( "destroy_after_turn" ? "destroy" : undefined; - await db.transaction(async (tx) => { + await options.beforeNativeRuntimeSelection?.(run.id); + const nativeSelected = await db.transaction(async (tx) => { const lockedRun = await tx .select() .from(heartbeatRuns) @@ -22663,6 +22669,14 @@ export function heartbeatService( .limit(1) .then((rows) => rows[0] ?? null); if (!lockedRun) throw new Error("native_runtime_run_missing"); + // Cancellation and runtime selection serialize on this row. A + // stopped preparation must never create a new native coordinator. + if (lockedRun.status !== "running" || lockedRun.resultJson?.startupCancellation) return false; + if (lockedRun.runtimeMode === "legacy" && lockedRun.controllerBootId && + !(await renewLegacyControllerLease(tx as unknown as Db, lockedRun))) { + nativeOwnershipHeld = true; + return false; + } if ( lockedRun.runtimeModeResolvedAt && lockedRun.runtimeMode !== "native" @@ -22770,7 +22784,9 @@ export function heartbeatService( phase: "observed", }) .onConflictDoNothing(); + return true; }); + if (!nativeSelected) return; controllerLease.stop(); nativeWorkspaceSync = await prepareNativeWorkspaceSync({ db, @@ -23303,6 +23319,7 @@ export function heartbeatService( }), ); if (!guardedDispatch.dispatched) return; + nativeDispatchStarted = true; adapterResult = await guardedDispatch.resultPromise; } finally { await nativeGitHubBridge?.stop(); @@ -25063,6 +25080,17 @@ export function heartbeatService( }); } } + if (latestRun?.status === "cancelled" && !nativeDispatchStarted && !nativeOwnershipHeld && + (latestRun.runtimeMode === "native" || + parseObject(latestRun.resultJson?.startupCancellation).beforeNativeSelection === true)) { + // This executor has finished preparation and lease cleanup without + // handing off to native execution. Keep a durable receipt for admission + // after a restart; cleanup receipts are independently rechecked there. + await db.update(heartbeatRuns).set({ + resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || + ${JSON.stringify({ startupPreparationSettledAt: new Date().toISOString() })}::jsonb`, + }).where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.status, "cancelled"))); + } // Interrupting a queued message explicitly authorizes the pending queue. // Retry its normal promotion after leases and adapter cleanup have settled; // the earlier terminal write can still have an execution blocker here. @@ -27619,7 +27647,7 @@ export function heartbeatService( reason = "Cancelled by control plane", options: CancelRunOptions = {}, ) { - const run = await getRun(runId); + let run = await getRun(runId); if (!run) throw notFound("Heartbeat run not found"); const pendingNativeRetry = run.runtimeMode === "native" && run.status === "failed" @@ -27644,16 +27672,6 @@ export function heartbeatService( return run; const agent = await getAgent(run.agentId); const errorCode = options.errorCode ?? "cancelled"; - const resultJson = agent - ? { - ...mergeRunStopMetadataForAgent(agent, "cancelled", { - resultJson: parseObject(run.resultJson), - errorCode, - errorMessage: reason, - }), - ...(options.resultJson ?? {}), - } - : options.resultJson; const pendingProcessCancellation = processRunCancellationSettlements.get( run.id, @@ -27670,6 +27688,39 @@ export function heartbeatService( ? captureAdapterStopOwnership(run.id) : undefined; const control = stopOwnership?.control; + // Capture the existing adapter owner before waiting on the run lock. Then + // atomically fence preparation and refresh the selected runtime, so Stop + // cannot miss a native handoff that won after its first read. + // Established legacy processes must still be stopped if the database is + // unavailable. Only native or not-yet-dispatched preparation needs this + // additional durable fence before its existing cancellation path. + if (run.runtimeMode === "native" || (!run.runtimeModeResolvedAt && !running && !control)) { + const [fenced] = await db.update(heartbeatRuns).set({ + resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || + jsonb_build_object('startupCancellation', jsonb_build_object( + 'requestedAt', ${new Date().toISOString()}::text, + 'beforeNativeSelection', ${heartbeatRuns.runtimeMode} = 'legacy' + and ${heartbeatRuns.runtimeModeResolvedAt} is null + and ${heartbeatRuns.executionStage} = 'preparing' + and coalesce(${heartbeatRuns.runnerProfileJson}->'adapterDispatch'->>'adapterType' = 'paperclip_runner', false) + ))`, + }).where(and(eq(heartbeatRuns.id, runId), inArray(heartbeatRuns.status, + pendingNativeRetry ? [...CANCELLABLE_HEARTBEAT_RUN_STATUSES, "failed"] : [...CANCELLABLE_HEARTBEAT_RUN_STATUSES], + ))).returning(); + if (!fenced) return getRun(runId); + run = fenced; + } + const resultJson = agent + ? { + ...mergeRunStopMetadataForAgent(agent, "cancelled", { + resultJson: parseObject(run.resultJson), + errorCode, + errorMessage: reason, + }), + ...(options.resultJson ?? {}), + } + : options.resultJson; + try { let releaseProcessCancellation: (() => void) | undefined; const processCancellationSettlement = diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 08bd055daa..93b407e818 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -4146,6 +4146,7 @@ function leaseDb( runResultJson: Record = {}, updates: Array<{ table: unknown; values: Record }> = [], runnerProfileJson: Record = {}, + runStatus = "running", ): Db { const coordinator: LeaseCoordinator = { runId: boundExecution.binding.runId, @@ -4189,6 +4190,7 @@ function leaseDb( resultJson: runResultJson, runnerProfileJson, runtimeMode: "native", + status: runStatus, }, ] : table === issues @@ -6527,6 +6529,28 @@ describe("native process ownership", () => { ); }); + it.each(["cancelled", "succeeded", "interrupted", "timed_out", "failed"])( + "refuses native provider claims after the run became %s", async status => { + const updates: Array<{ table: unknown; values: Record }> = []; + state.createBackend.mockClear(); + await expect(executePaperclipNativeSession({ + db: leaseDb(execution, {}, {}, updates, {}, status), execution, runnerInstanceId: "late-startup", + })).rejects.toThrow(); + expect(state.createBackend).not.toHaveBeenCalled(); + expect(updates.some(update => update.table === nativeRunFinalizations)).toBe(false); + expect(updates.some(update => update.values.eventType === "native.process_start_requested")).toBe(false); + }, + ); + + it("fences a cancellation request before its terminal status commits", async () => { + state.createBackend.mockClear(); + await expect(executePaperclipNativeSession({ + db: leaseDb(execution, {}, { startupCancellation: { requestedAt: new Date().toISOString() } }), + execution, runnerInstanceId: "cancel-requested", + })).rejects.toThrow(); + expect(state.createBackend).not.toHaveBeenCalled(); + }); + it("forwards the app-server PID and process group through the production backend seam", async () => { const processMetadata = { pid: 42_001, diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index cab73abe1c..9a24cf3faa 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -6965,6 +6965,7 @@ async function executePaperclipNativeSessionWithinScope( nativeIssueId: heartbeatRuns.nativeIssueId, resultJson: heartbeatRuns.resultJson, runtimeMode: heartbeatRuns.runtimeMode, + status: heartbeatRuns.status, }) .from(heartbeatRuns) .where(eq(heartbeatRuns.id, input.execution.binding.runId)) @@ -6980,6 +6981,12 @@ async function executePaperclipNativeSessionWithinScope( ) { throw new Error("native_execution_binding_changed"); } + // A cancellation can win after heartbeat dispatch admission but + // before this claim. Never revive a terminal run or a settled startup. + if (boundRun.status !== "running" || boundRun.resultJson?.startupCancellation || + coordinator.phase === "terminal_failure") { + throw new NativeCancellationPendingRecoveryError(); + } const cancellationIntent = record( record(boundRun.resultJson).nativeCancellation, );