diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 336687a07a6..ec5752baa18 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2621,6 +2621,26 @@ configuration, and controller-owned credential or instruction resources use verified suspension and saved-history resume. This recovery can restart a provider; it does not suspend every healthy warm turn. +A different task can reuse the same physical sandbox after its prior task +finishes. Provider ingress has one listener per allocation. Before a new logical +session starts, Runner retires a locally retained idle session that still owns +that listener. The new run must hold the current native claim, task assignment, +environment lease and remote workspace writer. Company, agent, provider, +allocation, workspace and directory must match. The prior run must be committed +and accounted, with its original controller generation and a clean released or +expired lease. Retirement saves the prior session through its own checkpoint and +file-collection callbacks and waits for close before the new listener binds. +After close, admission rechecks the canonical native claim, task, workspace and +leases before staging files or invoking the new backend. Ownership loss, Stop +or a changed writer during retirement cannot authorize the new execution. +The new task keeps its own logical session and authority. + +Active owners, a changed controller or writer, missing canonical proof and failed +close block admission. A retained entry stays available for verified retirement; +no process is selected by its port number. This rule covers current-controller +warm entries with recorded physical identity. A historical process outside that +registry still needs the existing exact-owner recovery proof. + A fresh native session admitted for the current task may replace a terminal predecessor's goal source epoch when the saved goal is absent or complete. It must hold the task's execution lock and assignee. Its accepted terminal diff --git a/doc/testing-native-remote-continuity.md b/doc/testing-native-remote-continuity.md index bd4fb833711..810c16e9060 100644 --- a/doc/testing-native-remote-continuity.md +++ b/doc/testing-native-remote-continuity.md @@ -540,3 +540,26 @@ explicit incomplete usage report before failure; this preserves known budget exposure without settling ambiguous usage. That required report and its receipt retain the accounting measurements and matching saved prior identity, so an optional diagnostic acknowledgement cannot discard reconciliation evidence. + +## Shared allocation ingress retirement + +The `retires the completed child listener` cases in +`native-session-executor.test.ts` use the actual executor and warm registry, +PostgreSQL ownership rows, and real local TCP listeners. The provider process +boundary is supplied by the fixture. The listeners use an ephemeral test port; +production provider ingress keeps its fixed port. A completed child's listener +must close before a distinct parent session binds in that allocation. The child's +accepted result stays unchanged. + +```sh +pnpm --filter @paperclipai/server exec vitest run \ + src/services/native-runtime/native-session-executor.test.ts \ + -t 'retires the completed child listener' +``` + +Controls cover foreign companies, allocations and workspaces; live or unaccounted +owners; changed controllers, reassignment and competing leases; workspace-lock +and cancellation races; unconfirmed close; and controller, Stop and writer +changes inside awaited close. The latter must prevent parent backend invocation +even when the old listener has closed. These local checks do not identify +a hosted conflicting PID or prove a hosted provider Stop or recovery. diff --git a/server/src/__tests__/helpers/native-warm-ingress-fixture.ts b/server/src/__tests__/helpers/native-warm-ingress-fixture.ts new file mode 100644 index 00000000000..15f904de509 --- /dev/null +++ b/server/src/__tests__/helpers/native-warm-ingress-fixture.ts @@ -0,0 +1,64 @@ +import { randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import * as tables from "@paperclipai/db"; +import type { NativeExecutionInputV1, NativeExecutionInputV3 } from "@paperclipai/paperclip-runner"; +import { startRecoveryPostgres } from "./recovery-postgres.js"; +import { prepareNativeHeartbeatRun } from "../../services/native-runtime/prepare-native-run.js"; +import { buildNativeCompletionContract } from "../../services/native-runtime/completion-contracts.js"; +import { nativeRuntimeContextFixture } from "../../services/native-runtime/runtime-context.test-fixture.js"; + +export async function createNativeWarmIngressFixture(base: NativeExecutionInputV1, cwd: string) { + const database = await startRecoveryPostgres("shared-ingress-owner-"); + const db = tables.createDb(database.connectionString); + const companyId = randomUUID(), agentId = randomUUID(), environmentId = randomUUID(), workspaceId = randomUUID(); + const providerLeaseId = `allocation-${randomUUID()}`; + await db.insert(tables.companies).values({ id: companyId, name: "Shared ingress", issuePrefix: `I${companyId.slice(0, 7)}` }); + await db.insert(tables.agents).values({ id: agentId, companyId, name: "Shared worker", adapterType: "paperclip_runner" }); + await db.insert(tables.environments).values({ id: environmentId, name: "Shared allocation", driver: "sandbox" }); + await db.insert(tables.executionWorkspaces).values({ id: workspaceId, companyId, name: "Shared workspace", mode: "shared_workspace", + strategyType: "project_primary", providerType: "cloud_sandbox", cwd }); + const defaultCompanyId = companyId, defaultAgentId = agentId, defaultWorkspaceId = workspaceId, defaultProviderLeaseId = providerLeaseId; + let epoch = 0; + async function admit(options?: { companyId?: string; agentId?: string; workspaceId?: string; providerLeaseId?: string }) { + const ownerCompanyId = options?.companyId ?? defaultCompanyId, ownerAgentId = options?.agentId ?? defaultAgentId; + const ownerWorkspaceId = options?.workspaceId ?? defaultWorkspaceId, ownerProviderLeaseId = options?.providerLeaseId ?? defaultProviderLeaseId; + const companyId = ownerCompanyId, agentId = ownerAgentId, workspaceId = ownerWorkspaceId, providerLeaseId = ownerProviderLeaseId; + const issueId = randomUUID(), runId = randomUUID(), leaseId = randomUUID(); + const [issue] = await db.insert(tables.issues).values({ id: issueId, companyId, title: "Shared handoff", status: "in_progress", + assigneeAgentId: agentId, executionWorkspaceId: workspaceId }).returning(); + const [run] = await db.insert(tables.heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: issueId, + status: "running", runtimeMode: "native", contextSnapshot: { issueId } }).returning(); + await db.update(tables.issues).set({ executionRunId: runId }).where(eq(tables.issues.id, issueId)); + await db.insert(tables.environmentLeases).values({ id: leaseId, companyId, environmentId, executionWorkspaceId: workspaceId, + issueId, heartbeatRunId: runId, provider: "daytona", providerLeaseId, leasePolicy: "reuse_by_environment" }); + await db.insert(tables.nativeRunFinalizations).values({ runId, companyId, issueId, phase: "observed" }); + await prepareNativeHeartbeatRun({ db, run, issue, environmentLeaseId: leaseId }); + const [contract] = await db.select().from(tables.completionContracts).where(eq(tables.completionContracts.issueId, issueId)); + const execution: NativeExecutionInputV3 = { ...base, schema: "paperclip.native-execution-input.v3", executionMode: "default", planningContext: null, + runtimeContext: nativeRuntimeContextFixture(), binding: { companyId, agentId, issueId, runId, executionWorkspaceId: workspaceId }, + completionContract: { id: contract.id, sha256: contract.canonicalSha256, schemaVersion: "paperclip.completion-contract.v1", + contract: buildNativeCompletionContract(issue, { revision: contract.revision }) }, + workspace: { cwd, repoUrl: null, repoRef: null, branchName: null }, + session: { ...base.session, normalizedSessionId: `logical-${issueId}`, lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 } } }; + await db.update(tables.executionWorkspaces).set({ metadata: { remoteWorkspace: { + schema: "paperclip.remote-workspace.v1", epoch: ++epoch, snapshotId: null, source: { kind: "empty" }, writerRunId: runId, dirty: true, + location: { environmentId, leaseId, providerLeaseId, remoteCwd: cwd }, + } } }).where(eq(tables.executionWorkspaces.id, workspaceId)); + const target = { kind: "remote" as const, transport: "sandbox" as const, providerKey: "daytona", environmentId, leaseId, remoteCwd: cwd, + effectiveCapabilities: { runnerWebSocketIngress: true } as never, + sandboxLeaseAcquisition: { outcome: "resumed" as const, providerLeaseId } }; + return { execution, target, issueId, runId, leaseId }; + } + async function settle(owner: Awaited>, result: Record) { + const resultId = randomUUID(); + await db.insert(tables.nativeRunResults).values({ id: resultId, companyId, issueId: owner.issueId, runId: owner.runId, + completionContractId: owner.execution.completionContract.id, schemaStatus: "accepted", resultJson: result, + serverFingerprint: randomUUID(), canonicalSha256: randomUUID() }); + await db.update(tables.nativeRunFinalizations).set({ phase: "committed", resultId, leaseOwner: null, leaseExpiresAt: null }).where(eq(tables.nativeRunFinalizations.runId, owner.runId)); + await db.update(tables.heartbeatRuns).set({ status: "succeeded", finishedAt: new Date(), nativePhase: "committed", costAccountingPending: false }).where(eq(tables.heartbeatRuns.id, owner.runId)); + await db.update(tables.issues).set({ status: "done", executionRunId: null, checkoutRunId: null }).where(eq(tables.issues.id, owner.issueId)); + await db.update(tables.environmentLeases).set({ status: "expired", cleanupStatus: "success", releasedAt: new Date() }).where(eq(tables.environmentLeases.id, owner.leaseId)); + return resultId; + } + return { db, companyId, agentId, environmentId, workspaceId, providerLeaseId, admit, settle, cleanup: () => database.cleanup() }; +} 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 ef43bf02e7f..e3611e34991 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -10011,6 +10011,138 @@ describe("native warm session supervision", () => { expect(state.execute).toHaveBeenCalledTimes(2); }); + it.each(["completed", "different-allocation", "foreign-company", "different-workspace", "live-owner", "unaccounted-owner", "changed-controller", "forged-allocation", "foreign-writer", "foreign-active-lease", "reassigned-child", "parent-already-started", "writer-lock-race", "stop-after-claim", "controller-change-during-close", "stop-during-close", "writer-change-during-close", "retirement-fails", "foreign-logical-session", "stopped-parent"] as const)("retires the completed child listener before a different parent task uses the same sandbox: %s", async scenario => { + const { createServer } = await import("node:net"); + const { createNativeWarmIngressFixture } = await import("../../__tests__/helpers/native-warm-ingress-fixture.js"); + const { eq } = await import("drizzle-orm"); + const tables = await import("@paperclipai/db"); + const f = await createNativeWarmIngressFixture(execution, join(isolatedHome, "shared-workspace")); + await mkdir(join(isolatedHome, "shared-workspace")); + const listener = createServer(); + await new Promise((resolve, reject) => { listener.once("error", reject); listener.listen(0, "127.0.0.1", resolve); }); + const port = (listener.address() as import("node:net").AddressInfo).port; + let failClose = scenario === "retirement-fails"; + let unlockWriter: (() => void) | undefined; + let writerTransaction: Promise | undefined; + let incomingRunId: string | undefined; + const close = vi.fn(async () => { + if (failClose) throw new Error("owned close is unconfirmed"); + if (scenario === "controller-change-during-close" && incomingRunId) await f.db.update(nativeRunFinalizations) + .set({ leaseOwner: "replacement-owner", controllerBootId: "replacement-controller", controllerGeneration: 999 }) + .where(eq(nativeRunFinalizations.runId, incomingRunId)); + if (scenario === "stop-during-close" && incomingRunId) await f.db.update(heartbeatRuns) + .set({ status: "cancelled" }).where(eq(heartbeatRuns.id, incomingRunId)); + if (scenario === "writer-change-during-close" && incomingRunId) { + const [workspace] = await f.db.select().from(tables.executionWorkspaces).where(eq(tables.executionWorkspaces.id, f.workspaceId)); + await f.db.update(tables.executionWorkspaces).set({ metadata: { ...workspace.metadata, + remoteWorkspace: { ...(workspace.metadata!.remoteWorkspace as Record), writerRunId: randomUUID() } } }) + .where(eq(tables.executionWorkspaces.id, f.workspaceId)); + } + if (listener.listening) await new Promise((resolve, reject) => listener.close(error => error ? reject(error) : resolve())); + }); + state.execute.mockReset().mockImplementationOnce(async options => { + await options.onSession?.({ close }); return result; + }).mockImplementationOnce(async options => { + expect(options.existingSession).toBeUndefined(); + const parentListener = createServer(); + try { + await new Promise((resolve, reject) => { parentListener.once("error", reject); parentListener.listen(scenario === "different-allocation" ? 0 : port, "127.0.0.1", resolve); }); + } finally { + if (parentListener.listening) await new Promise(resolve => parentListener.close(() => resolve())); + } + return result; + }); + try { + const child = await f.admit(); + await executePaperclipNativeSession({ db: f.db, execution: child.execution, runnerInstanceId: "child-runner", useRunnerd: true, + backend: {} as never, runnerExecutionTarget: child.target }); + const resultId = await f.settle(child, result); + const options: Parameters[0] = {}; + if (scenario === "different-allocation") options.providerLeaseId = `different-${f.providerLeaseId}`; + if (scenario === "different-workspace" || scenario === "foreign-company") { + options.workspaceId = randomUUID(); + if (scenario === "foreign-company") { + options.companyId = randomUUID(); options.agentId = randomUUID(); + await f.db.insert(tables.companies).values({ id: options.companyId, name: "Foreign tenant", issuePrefix: `F${options.companyId.slice(0, 7)}` }); + await f.db.insert(tables.agents).values({ id: options.agentId, companyId: options.companyId, name: "Foreign worker", adapterType: "paperclip_runner" }); + } + await f.db.insert(tables.executionWorkspaces).values({ id: options.workspaceId, companyId: options.companyId ?? f.companyId, + name: "Other workspace", mode: "shared_workspace", strategyType: "project_primary", providerType: "cloud_sandbox", cwd: child.execution.workspace.cwd }); + } + const parent = await f.admit(options); + incomingRunId = parent.runId; + if (scenario === "live-owner") await f.db.update(heartbeatRuns).set({ status: "running", nativePhase: "observed", finishedAt: null }).where(eq(heartbeatRuns.id, child.runId)); + if (scenario === "unaccounted-owner") await f.db.update(heartbeatRuns).set({ costAccountingPending: true }).where(eq(heartbeatRuns.id, child.runId)); + if (scenario === "changed-controller") await f.db.update(nativeRunFinalizations).set({ controllerBootId: "foreign-controller" }).where(eq(nativeRunFinalizations.runId, child.runId)); + if (scenario === "forged-allocation") await f.db.update(environmentLeases).set({ providerLeaseId: "canonical-other-allocation" }).where(eq(environmentLeases.id, parent.leaseId)); + if (scenario === "foreign-writer") { + const [workspace] = await f.db.select().from(tables.executionWorkspaces).where(eq(tables.executionWorkspaces.id, f.workspaceId)); + await f.db.update(tables.executionWorkspaces).set({ metadata: { ...workspace.metadata, + remoteWorkspace: { ...(workspace.metadata!.remoteWorkspace as Record), writerRunId: randomUUID() } } }).where(eq(tables.executionWorkspaces.id, f.workspaceId)); + } + if (scenario === "reassigned-child") await f.db.update(tables.issues).set({ assigneeAgentId: null }).where(eq(tables.issues.id, child.issueId)); + if (scenario === "parent-already-started") await f.db.update(heartbeatRuns).set({ processPid: 2147483000 }).where(eq(heartbeatRuns.id, parent.runId)); + if (scenario === "foreign-active-lease") { + const issueId = randomUUID(), runId = randomUUID(); + await f.db.insert(tables.issues).values({ id: issueId, companyId: f.companyId, title: "Live sibling", assigneeAgentId: f.agentId, status: "in_progress" }); + await f.db.insert(heartbeatRuns).values({ id: runId, companyId: f.companyId, agentId: f.agentId, nativeIssueId: issueId, runtimeMode: "native", status: "running" }); + await f.db.insert(environmentLeases).values({ companyId: f.companyId, environmentId: f.environmentId, executionWorkspaceId: f.workspaceId, + issueId, heartbeatRunId: runId, provider: "daytona", providerLeaseId: f.providerLeaseId, status: "active" }); + } + if (scenario === "foreign-logical-session") parent.execution.session.normalizedSessionId = child.execution.session.normalizedSessionId; + if (scenario === "stopped-parent") await f.db.update(heartbeatRuns).set({ status: "cancelled" }).where(eq(heartbeatRuns.id, parent.runId)); + let claimed!: () => void; + const claimObserved = new Promise(resolve => { claimed = resolve; }); + if (scenario === "writer-lock-race") { + let locked!: () => void; + const lockObserved = new Promise(resolve => { locked = resolve; }); + const release = new Promise(resolve => { unlockWriter = resolve; }); + writerTransaction = f.db.transaction(async tx => { + const [workspace] = await tx.select().from(tables.executionWorkspaces).where(eq(tables.executionWorkspaces.id, f.workspaceId)).for("update"); + locked(); await release; + await tx.update(tables.executionWorkspaces).set({ metadata: { ...workspace.metadata, + remoteWorkspace: { ...(workspace.metadata!.remoteWorkspace as Record), writerRunId: randomUUID() } } }).where(eq(tables.executionWorkspaces.id, f.workspaceId)); + }); + await lockObserved; + } + const start = executePaperclipNativeSession({ db: f.db, execution: parent.execution, runnerInstanceId: "parent-runner", useRunnerd: true, + backend: {} as never, runnerExecutionTarget: parent.target, onExecutionLeaseAcquired: async () => { + claimed(); + if (scenario === "stop-after-claim") await f.db.update(heartbeatRuns).set({ status: "cancelled" }).where(eq(heartbeatRuns.id, parent.runId)); + } }); + const outcome = start.then(value => ({ value }), error => ({ error })); + if (scenario === "writer-lock-race") { + await claimObserved; + expect(close).not.toHaveBeenCalled(); + unlockWriter!(); await writerTransaction; + } + if (scenario === "completed" || scenario === "different-allocation") { + await start; + expect(close).toHaveBeenCalledTimes(scenario === "completed" ? 1 : 0); + expect(state.execute).toHaveBeenCalledTimes(2); + const admittedParent = state.execute.mock.calls[1][0]; + expect(admittedParent.input.binding.issueId).toBe(parent.issueId); + expect(admittedParent.input.session.normalizedSessionId).toBe(parent.execution.session.normalizedSessionId); + expect(admittedParent.input.session.normalizedSessionId).not.toBe(child.execution.session.normalizedSessionId); + expect(admittedParent.existingSession).toBeUndefined(); + } else { + await expect(start).rejects.toThrow(); + expect(close).toHaveBeenCalledTimes(["retirement-fails", "controller-change-during-close", "stop-during-close", "writer-change-during-close"].includes(scenario) ? 1 : 0); + expect(listener.listening).toBe(!["controller-change-during-close", "stop-during-close", "writer-change-during-close"].includes(scenario)); + expect(state.execute).toHaveBeenCalledOnce(); + } + await outcome; + const [retained] = await f.db.select().from(nativeRunResults).where(eq(nativeRunResults.id, resultId)); + expect(retained.resultJson).toEqual(result); + } finally { + failClose = false; + unlockWriter?.(); await writerTransaction; + await closeWarmNativeSessionsForEnvironment({ environmentId: f.environmentId, reason: "shared ingress fixture cleanup" }); + if (listener.listening) await new Promise(resolve => listener.close(() => resolve())); + await f.cleanup(); + } + }, 60_000); + it("claims native execution authority before retiring a superseded warm identity", async () => { const { base, next, run } = fixture("identity-native-claim-order"); const updates: Array<{ table: unknown; values: Record }> = []; diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 6bbff7593c7..a2600506bd0 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -229,6 +229,7 @@ import { } from "./remote-runner-recovery-diagnostics.js"; import { readRemoteRunnerRecoveryProbe, REMOTE_RUNNER_RECOVERY_PROBE_BUDGET_MS } from "./remote-runner-recovery-probe.js"; import { inspectRemoteWarmSessionRecovery, isSettledRemoteWarmControlState, type RemoteWarmSessionRecovery } from "./remote-warm-session-recovery.js"; +import { assertNativeWarmIngressRetirement, nativeWarmIngressAllocation, nativeWarmIngressIdentity, type NativeWarmIngressClaim, type NativeWarmIngressOwner } from "./native-warm-ingress-ownership.js"; import { resolveNativeStagedAttachmentRecovery, type NativeStagedAttachmentRecovery } from "./native-staged-attachment-recovery.js"; import { emitStartupReuse, environmentStartupReuse } from "@paperclipai/adapter-utils/startup-reuse"; import { persistActivity, publishActivity } from "../activity-log.js"; @@ -582,6 +583,7 @@ export type NativeInstructionWorkingCopy = { }; type WarmNativeSession = { + ingressOwner?: NativeWarmIngressOwner; onReleased: () => void; sessionCheckpointPersisted?: (snapshot: PersistedNativeSession) => void; pluginTaskId?: string; @@ -1858,6 +1860,7 @@ const sessionToolAuthorityEpochs = new Map(); const initializingSessionToolAuthorities = new Set(); const executingRunnerdSessionScopes = new Map(); const executingNativeOwnerScopes = new Map(); +const executingNativeIngressAllocations = new Map(); export function nativeSessionScopeDigestIsBusy(digest: string): boolean { const scopes = [ @@ -1964,21 +1967,32 @@ async function retireSupersededWarmNativeSessions( ownerScope: string, nextSessionScope: string, nextSessionId: string, + ingress?: { db: Db; claim: NativeWarmIngressClaim }, ): Promise { for (const [scope, entry] of warmNativeSessions) { - if (scope === nextSessionScope || entry.ownerScope !== ownerScope) continue; - if (entry.busy || executingRunnerdSessionScopes.has(scope)) { + if (scope === nextSessionScope) continue; + const logicalOwner = entry.ownerScope === ownerScope; + const priorIngress = entry.ingressOwner; + const physicalOwner = ingress && priorIngress && nativeWarmIngressAllocation(priorIngress.ingress) === nativeWarmIngressAllocation(ingress.claim.ingress); + if (!logicalOwner && !physicalOwner) continue; + if (entry.busy || entry.preparingRunId || executingRunnerdSessionScopes.has(scope)) { throw new Error("native_session_supervisor_busy"); } + if (!logicalOwner) { + await assertNativeWarmIngressRetirement(ingress!.db, ingress!.claim, priorIngress!); + if (warmNativeSessions.get(scope) !== entry || entry.ingressOwner !== priorIngress + || entry.busy || entry.preparingRunId || executingRunnerdSessionScopes.has(scope)) throw new Error("native_session_supervisor_busy"); + } if (entry.idleTimer !== null) clearTimeout(entry.idleTimer); entry.idleTimer = null; entry.busy = true; - entry.closeOnReleaseReason = "warm native session identity changed"; + entry.closeOnReleaseReason = logicalOwner ? "warm native session identity changed" : "warm native ingress requires another session"; try { await closeWarmNativeSession(entry, entry.closeOnReleaseReason, { - discardUnsettledSession: entry.normalizedSessionId !== nextSessionId, + discardUnsettledSession: logicalOwner && entry.normalizedSessionId !== nextSessionId, }); forgetWarmNativeSession(scope, entry); + if (!logicalOwner) await assertNativeWarmIngressRetirement(ingress!.db, ingress!.claim, priorIngress!); } finally { // A failed close retains the owner and prevents launch. A later attempt // must retry retirement, never adopt this partially closed transport. @@ -8310,7 +8324,17 @@ export async function executePaperclipNativeSession(input: { input.runnerExecutionTarget?.environmentId ?? null, ); const ownerToken = Symbol("native execution owner"); + const physicalIngress = input.useRunnerd ? nativeWarmIngressIdentity(input.execution, input.runnerExecutionTarget) : undefined; + const physicalIngressKey = physicalIngress ? nativeWarmIngressAllocation(physicalIngress) : undefined; + let ownsPhysicalIngress = false; try { + if (physicalIngressKey) { + const retained = warmNativeSessions.get(nativeSessionScopeKey(input.execution)); + if (retained && retained.issueId !== input.execution.binding.issueId) throw new Error("native_session_supervisor_busy"); + if (executingNativeIngressAllocations.has(physicalIngressKey)) throw new Error("native_session_supervisor_busy"); + executingNativeIngressAllocations.set(physicalIngressKey, ownerToken); + ownsPhysicalIngress = true; + } // Reserve across session identities before retiring or staging anything. if (executingNativeOwnerScopes.has(ownerScope)) { throw new Error("native_session_supervisor_busy"); @@ -8361,7 +8385,8 @@ export async function executePaperclipNativeSession(input: { const workspaceRoot = input.execution.workspace.cwd; if (workspaceRoot === null) throw new Error("native_workspace_required"); const needsWarmRetirement = [...warmNativeSessions].some(([scope, entry]) => - scope !== sessionScopeId && entry.ownerScope === ownerScope); + scope !== sessionScopeId && (entry.ownerScope === ownerScope || physicalIngress && entry.ingressOwner + && nativeWarmIngressAllocation(entry.ingressOwner.ingress) === nativeWarmIngressAllocation(physicalIngress))); const prepareAttachments = async () => { const admissionSha256 = nativeSha256(input.execution); const targetKind = input.runnerExecutionTarget?.kind ?? "local"; @@ -8440,6 +8465,9 @@ export async function executePaperclipNativeSession(input: { startupUnsettled = !(await startup.executionContext.drain()); startup.executionContext.finish(); } + if (ownsPhysicalIngress && physicalIngressKey && executingNativeIngressAllocations.get(physicalIngressKey) === ownerToken) { + executingNativeIngressAllocations.delete(physicalIngressKey); + } if (executingNativeOwnerScopes.get(ownerScope) === ownerToken) { executingNativeOwnerScopes.delete(ownerScope); } @@ -8936,9 +8964,12 @@ async function executePaperclipNativeSessionWithinScope( // Claim current native authority before the old warm checkpoint writer // must prove its same-controller successor. Shared staging stays after // strict predecessor suspension; it cannot run on a failed claim/Stop. + const ingress = nativeWarmIngressIdentity(input.execution, input.runnerExecutionTarget); if (input.useRunnerd && !claimedStopCleanup) await retireSupersededWarmNativeSessions( nativeSessionOwnerScope(input.execution, input.runnerExecutionTarget?.environmentId ?? null), nativeSessionScopeKey(input.execution), nativeSessionKey(input.execution), + ingress ? { db: input.db, claim: { ...input.execution.binding, ingress, leaseOwner, + controllerBootId: controller.bootId, controllerGeneration } } : undefined, ); claimedStartupContext?.assertOpen(); if (prepareAfterClaim) input = await prepareAfterClaim(); @@ -10030,11 +10061,18 @@ async function executePaperclipNativeSessionWithinScope( if (existing.session !== session) { existing.sessionCheckpointPersisted = runnerdBackend?.sessionCheckpointPersisted; } + const ingress = input.useRunnerd ? nativeWarmIngressIdentity(input.execution, input.runnerExecutionTarget) : undefined; + existing.ingressOwner = ingress ? { ...input.execution.binding, ingress, + controllerBootId: controller.bootId, controllerGeneration } : undefined; existing.session = session; existing.runId = input.execution.binding.runId; existing.issueId = input.execution.binding.issueId; } else warmNativeSessions.set(warmSessionId, { + ingressOwner: (() => { + const ingress = input.useRunnerd ? nativeWarmIngressIdentity(input.execution, input.runnerExecutionTarget) : undefined; + return ingress ? { ...input.execution.binding, ingress, controllerBootId: controller.bootId, controllerGeneration } : undefined; + })(), sessionCheckpointPersisted: runnerdBackend?.sessionCheckpointPersisted, onReleased: () => releaseRetainedNativeSessionCleanup(input.db, { companyId: input.execution.binding.companyId, agentId: input.execution.binding.agentId, diff --git a/server/src/services/native-runtime/native-warm-ingress-ownership.ts b/server/src/services/native-runtime/native-warm-ingress-ownership.ts new file mode 100644 index 00000000000..8a046006633 --- /dev/null +++ b/server/src/services/native-runtime/native-warm-ingress-ownership.ts @@ -0,0 +1,99 @@ +import { and, eq, inArray } from "drizzle-orm"; +import { environmentLeases, executionWorkspaces, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { readRemoteWorkspaceAuthority } from "@paperclipai/shared"; +import type { AdapterExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; +import type { NativeExecutionInput } from "../../vendor/paperclip-runner/index.js"; + +export type NativeWarmIngressIdentity = { + provider: string; + environmentId: string; + providerLeaseId: string; + leaseId: string; + workspaceId: string; + cwd: string; +}; +export type NativeWarmIngressOwner = { + companyId: string; + agentId: string; + issueId: string; + runId: string; + controllerBootId: string; + controllerGeneration: number; + ingress: NativeWarmIngressIdentity; +}; +export type NativeWarmIngressClaim = NativeWarmIngressOwner & { leaseOwner: string }; + +export function nativeWarmIngressIdentity(execution: NativeExecutionInput, target?: AdapterExecutionTarget | null): NativeWarmIngressIdentity | undefined { + if (target?.kind !== "remote" || target.transport !== "sandbox" || target.effectiveCapabilities?.runnerWebSocketIngress !== true + || !target.providerKey || !target.environmentId || !target.leaseId || !target.sandboxLeaseAcquisition?.providerLeaseId + || !("cwd" in execution.workspace) || !execution.workspace.cwd + || execution.binding.executionWorkspaceId === execution.binding.runId) return undefined; + return { provider: target.providerKey, environmentId: target.environmentId, + providerLeaseId: target.sandboxLeaseAcquisition.providerLeaseId, leaseId: target.leaseId, + workspaceId: execution.binding.executionWorkspaceId, cwd: execution.workspace.cwd }; +} + +export function nativeWarmIngressAllocation(identity: NativeWarmIngressIdentity): string { + return JSON.stringify([identity.provider, identity.environmentId, identity.providerLeaseId]); +} + +const busy = () => new Error("native_session_supervisor_busy"); +export async function assertNativeWarmIngressRetirement(db: Db, current: NativeWarmIngressClaim, previous: NativeWarmIngressOwner): Promise { + if (current.companyId !== previous.companyId || current.agentId !== previous.agentId + || current.ingress.workspaceId !== previous.ingress.workspaceId || current.ingress.cwd !== previous.ingress.cwd + || nativeWarmIngressAllocation(current.ingress) !== nativeWarmIngressAllocation(previous.ingress) + || current.runId === previous.runId) throw busy(); + await db.transaction(async tx => { + const [workspace] = await tx.select().from(executionWorkspaces).where(and( + eq(executionWorkspaces.companyId, current.companyId), eq(executionWorkspaces.id, current.ingress.workspaceId), + )).for("update"); + const authority = readRemoteWorkspaceAuthority(workspace?.metadata); + if (!workspace || workspace.status !== "active" || authority?.writerRunId !== current.runId + || authority.location?.environmentId !== current.ingress.environmentId + || authority.location.providerLeaseId !== current.ingress.providerLeaseId + || authority.location.leaseId !== current.ingress.leaseId || authority.location.remoteCwd !== current.ingress.cwd) throw busy(); + const runs = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, current.companyId), + inArray(heartbeatRuns.id, [current.runId, previous.runId]))); + const incoming = runs.find(run => run.id === current.runId), prior = runs.find(run => run.id === previous.runId); + const coordinators = await tx.select().from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, current.companyId), + inArray(nativeRunFinalizations.runId, [current.runId, previous.runId]))); + const incomingCoordinator = coordinators.find(row => row.runId === current.runId), priorCoordinator = coordinators.find(row => row.runId === previous.runId); + const tasks = await tx.select().from(issues).where(and(eq(issues.companyId, current.companyId), inArray(issues.id, [current.issueId, previous.issueId]))); + const incomingTask = tasks.find(task => task.id === current.issueId), priorTask = tasks.find(task => task.id === previous.issueId); + const leases = await tx.select().from(environmentLeases).where(and(eq(environmentLeases.companyId, current.companyId), + inArray(environmentLeases.id, [current.ingress.leaseId, previous.ingress.leaseId]))); + const incomingLease = leases.find(lease => lease.id === current.ingress.leaseId), priorLease = leases.find(lease => lease.id === previous.ingress.leaseId); + const now = Date.now(); + if (!incoming || incoming.runtimeMode !== "native" || incoming.status !== "running" || incoming.nativePhase !== "observed" + || incoming.processPid !== null || incoming.processGroupId !== null || incoming.agentId !== current.agentId + || incoming.nativeIssueId !== current.issueId || incoming.resultJson?.startupCancellation || incoming.resultJson?.nativeCancellation + || incomingCoordinator?.issueId !== current.issueId || incomingCoordinator.phase !== "observed" || incomingCoordinator.resultId !== null || incomingCoordinator.leaseOwner !== current.leaseOwner + || incomingCoordinator.controllerBootId !== current.controllerBootId || incomingCoordinator.controllerGeneration !== current.controllerGeneration + || !incomingCoordinator.leaseExpiresAt || incomingCoordinator.leaseExpiresAt.getTime() <= now + || incomingTask?.assigneeAgentId !== current.agentId || incomingTask.executionRunId !== current.runId + || incomingTask.executionWorkspaceId !== current.ingress.workspaceId + || !["in_progress", "in_review"].includes(incomingTask.status) + || (incomingTask.checkoutRunId !== null && incomingTask.checkoutRunId !== current.runId) + || !prior || prior.runtimeMode !== "native" || prior.status !== "succeeded" || !prior.finishedAt || prior.costAccountingPending + || prior.agentId !== previous.agentId || prior.nativeIssueId !== previous.issueId || prior.nativePhase !== "committed" + || priorCoordinator?.issueId !== previous.issueId || priorCoordinator.phase !== "committed" || !priorCoordinator.resultId + || priorCoordinator.leaseOwner !== null || priorCoordinator.leaseExpiresAt !== null + || priorCoordinator.controllerBootId !== previous.controllerBootId || priorCoordinator.controllerGeneration !== previous.controllerGeneration + || !priorTask || priorTask.assigneeAgentId !== previous.agentId || priorTask.executionWorkspaceId !== previous.ingress.workspaceId + || priorTask.executionRunId !== null || priorTask.checkoutRunId !== null) throw busy(); + const leaseMatches = (lease: typeof environmentLeases.$inferSelect | undefined, owner: NativeWarmIngressOwner) => lease + && lease.heartbeatRunId === owner.runId && lease.issueId === owner.issueId + && lease.environmentId === owner.ingress.environmentId && lease.provider === owner.ingress.provider + && lease.providerLeaseId === owner.ingress.providerLeaseId && lease.executionWorkspaceId === owner.ingress.workspaceId; + if (!leaseMatches(incomingLease, current) || incomingLease!.status !== "active" + || (incomingLease!.expiresAt !== null && incomingLease!.expiresAt.getTime() <= now) + || !leaseMatches(priorLease, previous) || !["released", "expired"].includes(priorLease!.status) + || priorLease!.cleanupStatus !== "success") throw busy(); + const otherLeases = await tx.select({ runId: environmentLeases.heartbeatRunId }).from(environmentLeases).where(and( + eq(environmentLeases.companyId, current.companyId), eq(environmentLeases.environmentId, current.ingress.environmentId), + eq(environmentLeases.provider, current.ingress.provider), eq(environmentLeases.providerLeaseId, current.ingress.providerLeaseId), + eq(environmentLeases.status, "active"), + )); + if (otherLeases.some(lease => lease.runId !== current.runId)) throw busy(); + }); +}