diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 94bf4f37477..fc5dedd5249 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -3526,6 +3526,25 @@ reason uses only fixed lists of known names: command types, command statuses, owner phases, recovery states, lifecycle states, and fence reasons. Any other stored value appears as `other`. The reason never includes payloads or paths. +Live adoption of a warm Runner on a resumed sandbox names its failed proof as +`warm_recovery_refused=`, for example `journal_unsettled`, +`owner_changed` or `current_lease_changed`. After the predecessor's run row +proves `succeeded` and its coordinator proves `committed` with an accepted +result and no lease, such a refusal does not fail the task. The new run did no +provider work, so it fails with `native_predecessor_retirement_unproven` and +its next attempt starts on a fresh sandbox restored from the saved workspace, +under the same safety rules as any unretired predecessor. This covers a server +that was killed after a run parked on a question: its warm Runner stays `ready` +in the sandbox, and only the killed server could have adopted it. A refusal +because the new run itself is no longer admitted (stopped, reassigned, paused +or no longer owned by this controller) stays an ordinary failure. + +A follow-up on a warm session starts with its predecessor's Runner process +identity. That inherited identity alone does not count as a process of the +follow-up when it proves that it left the workspace unchanged. The identity +must equal a settled predecessor's identity of the same session, and the +follow-up must have no `native.process_identity_recorded` event of its own. + Resume source selection skips a failed follow-up that never started provider work. Such a run holds only the provider checkpoint that it inherited. Without the skip, a retry selects it, and controller recovery fails with diff --git a/server/src/services/native-runtime/native-controller-recovery.integration.test.ts b/server/src/services/native-runtime/native-controller-recovery.integration.test.ts index 194784bf039..b682b02f785 100644 --- a/server/src/services/native-runtime/native-controller-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-controller-recovery.integration.test.ts @@ -18,7 +18,8 @@ import { prepareNativeHeartbeatRun } from "./prepare-native-run.js"; import { nativeRunnerCheckpointWriter } from "./native-runner-checkpoint.js"; import { assertNativeRetainedSandboxBinding, createNativeControllerRecovery, deferNativeControllerRecovery, deferNativeRunForBusyAgentDirectory, NATIVE_AGENT_DIRECTORY_BUSY_WAIT, importNativeControllerRecovery, NativeControllerRecoveryWaitingError, NATIVE_RETAINED_OWNER_WAIT, resolveNativeControllerRecovery } from "./native-controller-recovery.js"; import { claimNativeSessionResumptions } from "./native-finalization-reconciler.js"; -import { claimNativeRestartRecoveries } from "./native-restart-recovery.js"; +import { claimNativeRestartRecoveries, currentNativeControllerIdentity } from "./native-restart-recovery.js"; +import { NativePredecessorRetirementUnprovenError } from "./predecessor-retirement.js"; import { AgentDirectoryPreparationDeferral, deferRunForBusyAgentDirectory, prepareWithAgentDirectoryDeferral } from "../heartbeat/agent-directory-busy.js"; import { AgentDirectoryOperationBusyError } from "../agent-directory-working-copies.js"; import { readNativeControllerRecoveryInventory } from "./native-controller-recovery-inventory.js"; @@ -521,6 +522,83 @@ describe.skipIf(!recoveryPostgresSupport.supported)("native controller durable d expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "duplicate", runIds: [runId], now: deadline })).toEqual([]); }); + it("sends a re-wake after a crash restart to a fresh sandbox when the parked predecessor's warm Runner cannot be adopted", async () => { + const fixture = await seed(), runId = randomUUID(), environmentId = randomUUID(), leaseId = randomUUID(); + const previous = parseNativeExecutionInput({ + schema: "paperclip.native-execution-input.v1", provider: { kind: "codex", model: null }, + binding: { companyId, agentId, issueId: fixture.scope.issueId, runId: fixture.scope.runId, executionWorkspaceId: randomUUID() }, + task: { identifier: "NCD-1", title: "Answer the parked question", description: null, + prompt: "Continue after the answer.", workMode: "standard" }, + workspace: { cwd: "/workspace", repoUrl: null, repoRef: null, branchName: null }, + session: { normalizedSessionId: fixture.scope.normalizedSessionId, driverKind: "codex_app_server", + protocolVersion: 1, lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 } }, + completionContract: { id: randomUUID(), sha256: "fixture", schemaVersion: "paperclip.completion-contract.v1", + contract: { revision: "1", objective: "Continue after the answer.", criteria: [{ id: "objective", requirement: "Save the result." }] } }, + credentialBindings: [], interactionResponses: [], + }); + const current = { ...previous, binding: { ...previous.binding, runId } }; + const checkpoint: PersistedNativeSession = { ...fixture.snapshot, driverKind: "codex_app_server", + providerSessionId: "parked-provider-session", activeTurnId: null, pendingRuntimeRequests: [] }; + const metadata = createNativeControllerRecovery(checkpoint, compatibility, { runId: fixture.scope.runId, + runnerInstanceId: fixture.native.runnerInstanceId, normalizedSessionId: fixture.scope.normalizedSessionId }); + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...fixture.profile, nativeExecutionInput: previous, + sessionCheckpoint: checkpoint, nativeControllerRecovery: metadata } }).where(eq(heartbeatRuns.id, fixture.scope.runId)); + await db.update(nativeRunFinalizations).set({ controllerPid: 2_147_483_000, controllerProcessStartedAt: new Date(Date.now() - 120_000) }) + .where(eq(nativeRunFinalizations.runId, fixture.scope.runId)); + const controller = await currentNativeControllerIdentity(); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: fixture.scope.issueId, + nativeSessionId: fixture.scope.normalizedSessionId, status: "running", runtimeMode: "native", + runnerProfileJson: { nativeExecutionInput: current } }); + await db.insert(nativeRunFinalizations).values({ companyId, issueId: fixture.scope.issueId, runId, phase: "observed", + leaseOwner: "replacement-controller", leaseExpiresAt: new Date(Date.now() + 60_000), + controllerBootId: controller.bootId, controllerPid: controller.pid }); + await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, fixture.scope.issueId)); + await db.insert(environments).values({ id: environmentId, name: `Parked ${environmentId}`, driver: "sandbox" }); + await db.insert(environmentLeases).values([ + { companyId, environmentId, issueId: fixture.scope.issueId, heartbeatRunId: fixture.scope.runId, provider: "daytona", + providerLeaseId: "warm-sandbox", status: "released", metadata: { remoteCwd: "/workspace" } }, + { id: leaseId, companyId, environmentId, issueId: fixture.scope.issueId, heartbeatRunId: runId, provider: "daytona", + providerLeaseId: "warm-sandbox", status: "active", metadata: { remoteCwd: "/workspace" } }, + ]); + const identity = { runId: fixture.scope.runId, normalizedSessionId: fixture.scope.normalizedSessionId, + runnerInstanceId: fixture.native.runnerInstanceId, environmentLeaseId: fixture.native.environmentLeaseId, + turnId: fixture.native.turnId, itemId: fixture.native.itemId }; + const parked = { identity, commands: [{ type: "turn.start", status: "completed" }], committedEvents: [ + { eventType: "semantic_tool.input", sourceEventId: "question", envelope: identity }, + { eventType: "run.terminal", envelope: identity }] }; + const inspect = (readControlState: () => Record = () => parked) => inspectRemoteWarmSessionRecovery({ + db, execution: current, identity, readControlState, + target: { kind: "remote", transport: "sandbox", environmentId, leaseId, providerKey: "daytona", remoteCwd: "/workspace", + sandboxLeaseAcquisition: { outcome: "resumed", providerLeaseId: "warm-sandbox" } } as never, + matchesExecution: saved => saved.binding.runId === fixture.scope.runId, + loadCheckpoint: () => checkpoint, canReuseProcess: () => true, + }); + const fallback = (error: unknown, reason: string) => { + expect(error).toBeInstanceOf(NativePredecessorRetirementUnprovenError); + expect(error).toMatchObject({ predecessorRunId: fixture.scope.runId, providerLeaseId: "warm-sandbox" }); + expect((error as Error).message).toContain(`warm_recovery_refused=${reason}`); + }; + + fallback(await inspect(() => ({ ...parked, commands: [...parked.commands, { type: "runner.suspend", status: "pending" }] })) + .catch(error => error), "journal_unsettled"); + + const recovered = (await inspect())!; + await expect(recovered.revalidate("replacement-controller")).resolves.toBeUndefined(); + await db.update(nativeRunFinalizations).set({ leaseOwner: "native-finalizer:replacement", leaseExpiresAt: new Date(Date.now() + 60_000) }) + .where(eq(nativeRunFinalizations.runId, fixture.scope.runId)); + fallback(await recovered.revalidate("replacement-controller").catch(error => error), "owner_changed"); + await db.update(nativeRunFinalizations).set({ leaseOwner: null, leaseExpiresAt: null }) + .where(eq(nativeRunFinalizations.runId, fixture.scope.runId)); + await db.update(environmentLeases).set({ status: "pending_cleanup" }).where(eq(environmentLeases.id, leaseId)); + fallback(await recovered.revalidate("replacement-controller").catch(error => error), "current_lease_changed"); + await db.update(environmentLeases).set({ status: "active" }).where(eq(environmentLeases.id, leaseId)); + + await db.update(heartbeatRuns).set({ status: "cancelled" }).where(eq(heartbeatRuns.id, runId)); + const stopped = await recovered.revalidate("replacement-controller").catch(error => error); + expect(stopped).not.toBeInstanceOf(NativePredecessorRetirementUnprovenError); + expect(stopped.message).toBe("runner_state_identity_mismatch: prior_owner_terminal_state_indeterminate; warm_recovery_refused=current_run_not_admitted"); + }); + // The Nov 10 staging shape: a preparation-only run's epoch was abandoned, so // the warm follow-up restored the last save into a new allocation of the same // sandbox. The committed source's Runner state stays in its own directory. diff --git a/server/src/services/native-runtime/remote-warm-session-recovery.ts b/server/src/services/native-runtime/remote-warm-session-recovery.ts index e2ce1bcbc9c..99d3d4207ea 100644 --- a/server/src/services/native-runtime/remote-warm-session-recovery.ts +++ b/server/src/services/native-runtime/remote-warm-session-recovery.ts @@ -4,6 +4,7 @@ import type { AdapterExecutionTarget } from "@paperclipai/adapter-utils/executio import { parseNativeExecutionInput, type NativeExecutionInput, type PersistedNativeSession, type DurableRecoveryIdentity } from "../../vendor/paperclip-runner/index.js"; import { currentNativeControllerIdentity, evaluateNativeControllerTakeover, recordLegacyHostFallbackProof } from "./native-restart-recovery.js"; import { NativeControllerRecoveryWaitingError, readNativeControllerRecovery } from "./native-controller-recovery.js"; +import { NativePredecessorRetirementUnprovenError } from "./predecessor-retirement.js"; export type RemoteWarmSessionRecovery = { identity: Record; @@ -114,6 +115,10 @@ export function describeFailedAttachment(failureMessage: unknown): string | null return `This run could not attach to its provider session: ${cause}. A later run may start a new provider thread. See ${SETTLEMENT_DOC}`; } +type WarmRecoveryRefusal = "identity_incomplete" | "target_not_resumed_sandbox" | "owner_unsettled" | "execution_mismatch" + | "checkpoint_unsettled" | "journal_unsettled" | "controller_not_stopped" | "prior_lease_mismatch" + | "current_run_not_admitted" | "current_owner_changed" | "owner_changed" | "current_lease_changed"; + export async function inspectRemoteWarmSessionRecovery(input: { db: Db; execution: NativeExecutionInput; @@ -124,13 +129,23 @@ export async function inspectRemoteWarmSessionRecovery(input: { loadCheckpoint: (previous: NativeExecutionInput) => PersistedNativeSession | null; canReuseProcess: (previous: NativeExecutionInput) => boolean; }): Promise { - const deny = (): never => { throw new Error("runner_state_identity_mismatch: prior_owner_terminal_state_indeterminate"); }; const { binding } = input.execution; const { target } = input; + let settledPredecessor: { runId: string; providerLeaseId: string } | null = null; + const deny = (reason: WarmRecoveryRefusal): never => { + const refusal = new Error(`runner_state_identity_mismatch: prior_owner_terminal_state_indeterminate; warm_recovery_refused=${reason}`); + if (settledPredecessor) { + throw new NativePredecessorRetirementUnprovenError(settledPredecessor.runId, settledPredecessor.providerLeaseId, refusal); + } + throw refusal; + }; + const refuseCurrentRun = (reason: WarmRecoveryRefusal): never => { + throw new Error(`runner_state_identity_mismatch: prior_owner_terminal_state_indeterminate; warm_recovery_refused=${reason}`); + }; const identity: Record = {}; for (const key of ["runId", "normalizedSessionId", "runnerInstanceId", "environmentLeaseId", "turnId", "itemId"]) { const value = input.identity[key]; - if (typeof value !== "string" || !value) return deny(); + if (typeof value !== "string" || !value) return deny("identity_incomplete"); identity[key] = value; } // A Runner lives in the sandbox it started in. A created or replacement @@ -142,28 +157,29 @@ export async function inspectRemoteWarmSessionRecovery(input: { } if (target.transport !== "sandbox" || !target.leaseId || !target.environmentId || !target.providerKey || !target.sandboxLeaseAcquisition || target.sandboxLeaseAcquisition.outcome !== "resumed" || - identity.runId === binding.runId) return deny(); + identity.runId === binding.runId) return deny("target_not_resumed_sandbox"); const providerLeaseId = target.sandboxLeaseAcquisition.providerLeaseId; const [prior] = await input.db.select({ run: heartbeatRuns, owner: nativeRunFinalizations }) .from(heartbeatRuns).innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id)) .where(and(eq(heartbeatRuns.id, identity.runId), eq(heartbeatRuns.companyId, binding.companyId), eq(heartbeatRuns.agentId, binding.agentId), eq(heartbeatRuns.nativeIssueId, binding.issueId))); if (!prior || prior.run.status !== "succeeded" || !prior.owner.resultId || !prior.owner.controllerPid || - !prior.owner.controllerProcessStartedAt || !prior.owner.controllerBootId) return deny(); + !prior.owner.controllerProcessStartedAt || !prior.owner.controllerBootId) return deny("owner_unsettled"); const previous = parseNativeExecutionInput(prior.run.runnerProfileJson?.nativeExecutionInput); - if (previous.binding.runId !== identity.runId || !input.matchesExecution(previous)) return deny(); + if (previous.binding.runId !== identity.runId || !input.matchesExecution(previous)) return deny("execution_mismatch"); if (prior.owner.leaseOwner && prior.owner.leaseExpiresAt && prior.owner.leaseExpiresAt > new Date()) { throw new NativeControllerRecoveryWaitingError(prior.owner.leaseExpiresAt); } - if (prior.owner.phase !== "committed" || prior.owner.leaseOwner) return deny(); + if (prior.owner.phase !== "committed" || prior.owner.leaseOwner) return deny("owner_unsettled"); + settledPredecessor = { runId: identity.runId, providerLeaseId }; const checkpoint = input.loadCheckpoint(previous); if (!checkpoint || checkpoint.identity.runId !== identity.runId || checkpoint.identity.companyId !== binding.companyId || checkpoint.identity.issueId !== binding.issueId || checkpoint.identity.agentId !== binding.agentId || checkpoint.identity.sessionId !== identity.normalizedSessionId || checkpoint.driverKind !== previous.session.driverKind || !checkpoint.providerSessionId || - checkpoint.activeTurnId !== null || checkpoint.pendingRuntimeRequests?.length !== 0) return deny(); + checkpoint.activeTurnId !== null || checkpoint.pendingRuntimeRequests?.length !== 0) return deny("checkpoint_unsettled"); const control = input.readControlState(); - if (!isSettledRemoteWarmControlState(control, identity)) return deny(); + if (!isSettledRemoteWarmControlState(control, identity)) return deny("journal_unsettled"); const assertPriorStopped = async () => { const persistedRecovery = readNativeControllerRecovery(prior.run.runnerProfileJson?.nativeControllerRecovery, prior.run.runnerProfileJson?.sessionCheckpoint); @@ -171,7 +187,7 @@ export async function inspectRemoteWarmSessionRecovery(input: { // durable controllers are fenced by the database lease and current turn. const stopped = await evaluateNativeControllerTakeover({ owner: prior.owner, now: new Date(), requireStoppedController: true, remoteLeaseFenced: persistedRecovery !== null }); - if (!stopped.allowed) return deny(); + if (!stopped.allowed) return deny("controller_not_stopped"); await recordLegacyHostFallbackProof(input.db, prior.run, stopped, "remote_warm_session_recovery"); }; await assertPriorStopped(); @@ -181,7 +197,7 @@ export async function inspectRemoteWarmSessionRecovery(input: { eq(environmentLeases.provider, target.providerKey), eq(environmentLeases.providerLeaseId, providerLeaseId), )); if (priorLeases.length !== 1 || priorLeases[0].metadata?.remoteCwd !== target.remoteCwd || - !["released", "expired", "active"].includes(priorLeases[0].status)) return deny(); + !["released", "expired", "active"].includes(priorLeases[0].status)) return deny("prior_lease_mismatch"); return { identity, checkpoint, previousExecution: previous, reuseProcess: input.canReuseProcess(previous), remote: { providerLeaseId, remoteCwd: target.remoteCwd }, @@ -206,15 +222,20 @@ export async function inspectRemoteWarmSessionRecovery(input: { if (!current || current.run.status !== "running" || current.run.resultJson?.nativeCancellation || current.company.status !== "active" || current.agent.lifecycleState !== "ready" || ["paused", "terminated", "pending_approval"].includes(current.agent.status) || - !priorOwner || priorOwner.leaseOwner || priorOwner.phase !== "committed" || - priorOwner.resultId !== prior.owner.resultId || priorOwner.controllerBootId !== prior.owner.controllerBootId || - current.owner.leaseOwner !== leaseOwner || !current.owner.leaseExpiresAt || - current.owner.leaseExpiresAt <= new Date() || current.owner.resultId || - current.owner.controllerBootId !== controller.bootId || current.owner.controllerPid !== controller.pid || current.issue.executionRunId !== binding.runId || current.issue.assigneeAgentId !== binding.agentId || - current.issue.status !== "in_progress" || !lease || lease.status !== "active" || + current.issue.status !== "in_progress") return refuseCurrentRun("current_run_not_admitted"); + if (current.owner.leaseOwner !== leaseOwner || !current.owner.leaseExpiresAt || + current.owner.leaseExpiresAt <= new Date() || current.owner.resultId || + current.owner.controllerBootId !== controller.bootId || current.owner.controllerPid !== controller.pid) { + return refuseCurrentRun("current_owner_changed"); + } + if (!priorOwner || priorOwner.leaseOwner || priorOwner.phase !== "committed" || + priorOwner.resultId !== prior.owner.resultId || priorOwner.controllerBootId !== prior.owner.controllerBootId) { + return deny("owner_changed"); + } + if (!lease || lease.status !== "active" || lease.environmentId !== target.environmentId || lease.provider !== target.providerKey || - lease.providerLeaseId !== providerLeaseId || lease.metadata?.remoteCwd !== target.remoteCwd) return deny(); + lease.providerLeaseId !== providerLeaseId || lease.metadata?.remoteCwd !== target.remoteCwd) return deny("current_lease_changed"); }, }; } diff --git a/server/src/services/remote-workspace-unchanged-proof.ts b/server/src/services/remote-workspace-unchanged-proof.ts index b02c08135a4..7c0fd82a5d1 100644 --- a/server/src/services/remote-workspace-unchanged-proof.ts +++ b/server/src/services/remote-workspace-unchanged-proof.ts @@ -6,6 +6,8 @@ import { import { remoteWorkspaceRunReferenceSchema, type RemoteWorkspaceAuthority } from "@paperclipai/shared"; import { executionStartedEventPredicate } from "./pre-dispatch-execution-evidence.js"; import { runOwnerFenceHeldSql } from "./remote-workspace-writer-hold.js"; +import { carriesInheritedRunnerIdentity } from "./native-runtime/native-inherited-runner-retry.js"; +import { PROCESS_IDENTITY_RECORDED } from "./native-local-process-stop.js"; /** * A writer's prepared files equal its saved version only while nothing ran in @@ -44,6 +46,15 @@ const FILE_NEUTRAL_OPERATION_PHASES = ["remote_workspace_restore", "workspace_fi /** Terminal states of a run that ended without a result. */ const ENDED_WITHOUT_RESULT = ["failed", "cancelled", "interrupted", "timed_out"]; +async function carriesOnlyInheritedRunnerIdentity(tx: Pick, + run: Parameters[1]): Promise { + if (!await carriesInheritedRunnerIdentity(tx, run)) return false; + const [own] = await tx.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), + eq(heartbeatRunEvents.eventType, PROCESS_IDENTITY_RECORDED))).limit(1); + return !own; +} + export type UnchangedProofRefusal = | "baseline_missing" | "authority_mismatch" | "writer_not_ended" | "writer_owner_live" | "process_recorded" | "reference_mismatch" | "provider_session_started" | "execution_recorded" | "result_recorded" @@ -68,15 +79,18 @@ export async function proveRemoteWorkspaceUnchanged(tx: Pick, inpu if (!authority.dirty || !authority.location || authority.epoch !== baseline.epoch || authority.snapshotId !== baseline.snapshotId) return { proven: false, reason: "authority_mismatch" }; - const [run] = await tx.select({ status: heartbeatRuns.status, processPid: heartbeatRuns.processPid, + const [run] = await tx.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, + nativeSessionId: heartbeatRuns.nativeSessionId, createdAt: heartbeatRuns.createdAt, + status: heartbeatRuns.status, processPid: heartbeatRuns.processPid, processGroupId: heartbeatRuns.processGroupId, processStartedAt: heartbeatRuns.processStartedAt, sessionIdAfter: heartbeatRuns.sessionIdAfter, runnerProfileJson: heartbeatRuns.runnerProfileJson, ownerFenceHeld: runOwnerFenceHeldSql() }).from(heartbeatRuns) .where(and(eq(heartbeatRuns.companyId, workspace.companyId), eq(heartbeatRuns.id, runId))); if (!run || !ENDED_WITHOUT_RESULT.includes(run.status)) return { proven: false, reason: "writer_not_ended" }; if (!input.callerOwnsWriter && run.ownerFenceHeld) return { proven: false, reason: "writer_owner_live" }; - if (run.processPid !== null || run.processGroupId !== null || run.processStartedAt !== null || run.sessionIdAfter !== null) - return { proven: false, reason: "process_recorded" }; + if (run.sessionIdAfter !== null) return { proven: false, reason: "process_recorded" }; + if ((run.processPid !== null || run.processGroupId !== null || run.processStartedAt !== null) + && !await carriesOnlyInheritedRunnerIdentity(tx, run)) return { proven: false, reason: "process_recorded" }; const profile = object(run.runnerProfileJson); const reference = remoteWorkspaceRunReferenceSchema.safeParse(profile.remoteWorkspace); if (profile.nativeWorkspaceSync != null || !reference.success || reference.data.state !== "prepared" diff --git a/server/src/services/remote-workspace-unchanged-release.test.ts b/server/src/services/remote-workspace-unchanged-release.test.ts index a218aee7266..736138c0c18 100644 --- a/server/src/services/remote-workspace-unchanged-release.test.ts +++ b/server/src/services/remote-workspace-unchanged-release.test.ts @@ -194,6 +194,24 @@ describe.skipIf(!support.supported)("remote workspace writer that ended without ? { dirty: false, writerRunId: null, snapshotId: f.savedA.snapshotId } : { dirty: true, writerRunId: f.runB }); }); + it.each([ + { label: "the settled predecessor's warm Runner identity", own: false, event: false, released: true }, + { label: "a Runner identity of its own", own: true, event: false, released: false }, + { label: "the predecessor's identity and its own identity record", own: false, event: true, released: false }, + ])("hands on the workspace of a follow-up that carries $label only when it never had its own Runner", async ({ own, event, released }) => { + const f = await preparedFollowUp(); + const nativeSessionId = randomUUID(), processStartedAt = new Date(Date.now() - 60_000); + await db.update(heartbeatRuns).set({ nativeSessionId, processPid: 41_001, processGroupId: 41_001, processStartedAt }) + .where(eq(heartbeatRuns.id, f.runA)); + await db.update(heartbeatRuns).set({ nativeSessionId, processPid: own ? 41_002 : 41_001, + processGroupId: own ? 41_002 : 41_001, processStartedAt }).where(eq(heartbeatRuns.id, f.runB)); + if (event) await f.event("native.process_identity_recorded"); + await f.end({ status: "failed", errorCode: "native_session_interrupted", controllerLeaseExpiresAt: new Date(Date.now() + 60_000) }); + expect(await releaseUnchangedFallbackWorkspace(db, { id: f.runB, companyId: f.companyId })).toBe(released); + expect(readRemoteWorkspaceAuthority((await f.workspace()).metadata)).toMatchObject(released + ? { dirty: false, writerRunId: null, snapshotId: f.savedA.snapshotId } : { dirty: true, writerRunId: f.runB }); + }); + it.each([1, 0])("queues no fallback attempt beside another writer (epoch change %i)", async epochChange => { const f = await preparedFollowUp(); await f.end({ status: "failed", errorCode: "adapter_failed" });