mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-04 19:28:46 +01:00
fix(native): retire idle shared-sandbox ingress owners before dispatch (#18893)
## Thinking Path > - Paperclip manages agents and their tasks. > - Native tasks can share one reusable sandbox workspace. > - A completed child can keep its provider process warm. > - A parent has its own logical session, but the sandbox has one ingress listener. > - Retirement by task identity leaves the child's listener in place. > - A new parent Runner then fails to bind that listener. > - This pull request retires a proven idle physical owner before the parent's distinct session starts. ## Linked Issues or Issue Description Refs #15353 and #17097 for listener configuration and warm transport rotation. Refs #18682 for owned process retirement. #18860 repairs a different resume-source failure. Searches found no existing repair for retained warm sessions on different tasks that share the same physical ingress allocation. **What happened?** A completed child and a later parent can use the same reusable sandbox. Warm-session retirement compares logical task ownership. It does not find the child's distinct task scope. The child keeps the fixed ingress listener, and the parent's new process fails with `runner_ingress_bind_conflict` / address already in use. **Expected behavior** A parent must start in its own logical session after the proven idle prior listener closes. Active, foreign or unverified owners must remain protected. **Steps to reproduce** Run the `retires the completed child listener` cases in `native-session-executor.test.ts`. The actual executor registers a warm child. PostgreSQL holds its committed, accounted result and clean expired lease. A distinct parent receives the same allocation and current workspace writer. The provider boundary uses real TCP listeners. The original executor reaches the second bind with the child still listening and fails with EADDRINUSE. The changed executor waits for the owned close, then starts the distinct parent session. **Paperclip version or commit** Reproduced onab2d2098a2. The branch includes current master44a4ca1c61before this PR was opened. **Deployment mode** Native Runner with provider ingress on a reusable sandbox workspace. Local regression uses real PostgreSQL and local TCP with a controlled provider-process boundary. No hosted provider call was made. ## What Changed - Keep physical ingress identity separate from logical task/session ownership. Never attach the parent to the child's session. - Reserve a physical allocation during local admission. Retire a distinct warm owner only after canonical company, agent, workspace, provider, allocation, lease, task and coordinator checks pass. - Require the current remote workspace writer and location to match the admitted run. Require the prior run to be committed and accounted, with its original controller generation and a clean released or expired lease. - Use the existing close, checkpoint and stopped-file collection path. Wait for close before parent dispatch, then recheck canonical authority before staging files or calling the parent backend. Retain the entry when close is uncertain. - Add actual executor/PostgreSQL/TCP safety cases and document the contract. Keep maps unchanged for this bug fix. ## Verification - Untouched production atab2d2098: actual executor + PostgreSQL + TCP fails EADDRINUSE. Same positive case passes after the change. - Final focused suite: 20 passed. It covers distinct parent session identity; foreign company, allocation and workspace; active or unaccounted owner; changed controller; reassignment; competing lease/writer; stale allocation metadata; Stop before and after claim; workspace-lock race; failed close that leaves the listener and accepted result intact; and controller transfer, Stop or writer change during awaited close. These last cases reject the stale parent before backend invocation after the old listener closes. - Untouched head3425acbe: transferring the incoming coordinator during close still reaches the parent backend. The same actual executor/PostgreSQL/TCP case passes after the post-close authority check. - Broader `native warm session supervision` suite: 160 passed. This run preceded the final two-line logical-session collision guard; the final 20-case suite includes that guard. - `pnpm --filter @paperclipai/server typecheck`: all source and test layers passed, including `src/__tests__`. - `pnpm --filter @paperclipai/server build`: production server and Runner build passed at current head572acfef79. The exact-head build stamp is recorded. - The provider/close boundary is controlled. The fixture seeds canonical committed/accounted finalization data in PostgreSQL. These tests do not claim a live model result, operator ACK, hosted PID-to-listener identity, or recovery of a historical task. - Full repository pnpm checks were not run locally for this server-only slice. Required current-head CI and reviews are pending. ## Risks The new retirement path can close a process from another task. It requires the same canonical company, agent, allocation and workspace, a current exclusive writer, and a completed idle owner. Missing or changed evidence leaves admission blocked. Active owners, changed generations, reassignment and cancellation remain fenced. A failed close retains the owner and saved result. This covers current-controller warm entries with recorded physical identity. Historical processes outside the registry keep the existing exact-owner recovery requirements. No port shuffle, port kill, task capability expansion, new polling loop or invented receipt is added. ## Model Used OpenAI GPT-6.1 Sol (`gpt-6.1-sol`) in Codex, with source editing, terminal execution and tool use. The host did not report its context-window size. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
d7283e8757
commit
e2dc2cc0be
6 files changed
+381
-5
No files matched your search
@@ -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
|
||||
|
||||
@@ -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.
|
||||
@@ -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<ReturnType<typeof admit>>, result: Record<string, unknown>) {
|
||||
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() };
|
||||
}
|
||||
@@ -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<void>((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<void> | 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<string, unknown>), writerRunId: randomUUID() } } })
|
||||
.where(eq(tables.executionWorkspaces.id, f.workspaceId));
|
||||
}
|
||||
if (listener.listening) await new Promise<void>((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<void>((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<void>(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<typeof f.admit>[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<string, unknown>), 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<void>(resolve => { claimed = resolve; });
|
||||
if (scenario === "writer-lock-race") {
|
||||
let locked!: () => void;
|
||||
const lockObserved = new Promise<void>(resolve => { locked = resolve; });
|
||||
const release = new Promise<void>(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<string, unknown>), 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<void>(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<string, unknown> }> = [];
|
||||
|
||||
@@ -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<string, SessionAuthorityEpoch>();
|
||||
const initializingSessionToolAuthorities = new Set<string>();
|
||||
const executingRunnerdSessionScopes = new Map<string, string>();
|
||||
const executingNativeOwnerScopes = new Map<string, symbol>();
|
||||
const executingNativeIngressAllocations = new Map<string, symbol>();
|
||||
|
||||
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<void> {
|
||||
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,
|
||||
|
||||
@@ -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<void> {
|
||||
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();
|
||||
});
|
||||
}
|
||||
Reference in new issue
Block a user