mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-08 01:57:31 +01:00
fix: retain native checkpoint ownership through warm close (#19225)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - A native warm session must save its checkpoint under its exact coordinator owner. > - A failed turn can return an accepted result before that checkpoint is saved. > - The executor stopped renewal and released the owner before the managed session closed. > - The backup owner guard then refused the uncommitted, released owner. A failed close could remain in the warm session map. > - This pull request keeps the claim and its existing renewal through checkpoint, close and instruction collection. > - The benefit is a valid checkpoint owner and one fenced release after the work joins. ## Linked Issues or Issue Description Refs #19208, #19206 and #14695. Those fixes cover a destroyed backup source, Stop before a provider session opens, and retained agent-file ownership. They do not change this checkpoint ownership order. **What happened?** A failed warm turn accepted its result. Its coordinator reached `workspace_finalizing`, with a result ID and no lease owner. Managed close then called the durable checkpoint writer. The writer correctly refused this state with `native_harness_storage_owner_missing`. The failed close could remain for the next fresh session to retire. **Expected behavior** The current claim and renewal remain valid until the required checkpoint, close and instruction collection join. Cleanup stops and joins renewal, then releases only that exact claim. A changed owner is not cleared. An earlier error keeps its identity. **Steps to reproduce** 1. Run the new real-Postgres regression against the base commit. 2. Complete a structured failed warm turn through the normal control-plane result producer. 3. Let the actual managed-session close call its production checkpoint callback. 4. Observe the missing-owner refusal after coordinator release. **Paperclip version or commit** Base: `56517dda6c`. The native ownership order also exists on the earlier failing revision. **Deployment mode** Native warm execution with durable harness storage and a managed remote workspace. ## What Changed - Join warm checkpoint, managed close and instruction collection before coordinator release. - Keep the existing lease renewal active through that work. Stop and join it before release. - Run the same fenced release after a checkpoint or collection error. Preserve the initiating error unless restart detachment has won. Keep the typed restart outcome and changed-owner fences. Attach only closed cleanup-failure flags. - Add real-Postgres ownership, delayed-renewal, error, retry and denial cases. Route the managed-session regression through the existing artifact-equipped native test lane. Update the continuity test for the retained claim. Document the active renewal lifetime. ## Verification - Red: the producing coordinator had an accepted result, `phase=workspace_finalizing` and `leaseOwner=null` at managed close. The production checkpoint callback refused it with `native_harness_storage_owner_missing`. - Green: all 9 real-Postgres regression cases pass. They cover a real 20-second renewal tick, a failed checkpoint, one force-fresh successor, and wrong-boot and changed-owner denial. Actual restart detachment wins even when checkpoint and joined renewal-stop both reject. The replacement claim remains unchanged. No new provider turn or cancellation receipt is created. A private canary does not enter the closed diagnostic flags or an error cause. - The executor, restart-recovery and heartbeat compatibility suites pass: 1,259 tests, with 20 gated skips. The container-enabled continuity case was not run locally. Its 21-second timer check remains. It now requires the exact claim during `hasChanges()` and checks release after the joined work. - All 39 shard-routing and combined-coverage tests pass. The new regression runs in the existing native lane with the required Runner artifact. - Final local head `0edcc37f35` passes `pnpm -r typecheck`, `pnpm build`, `node scripts/check-maps.mjs` and the 9-case regression. The earlier broader compatibility run used identical source bytes. A separate earlier repo-wide local suite on `6cdd68e915` ended with exit 1. Its general-server stage had one GitHub enrollment acknowledgement failure, 31,469 passing tests and 215 skipped tests. That acknowledgement failure reproduced in one narrow run on the same old tree. It is not an established flake or proven caused by this fix. Later repository stages did not run. No duplicate broad suite or full local pass is claimed. - The first published head failed review for detach-error precedence and the old continuity assertion. Its server shard also failed because the new test lacked a Runner daemon in the general lane. Those failures remain recorded. This candidate fixes their causes. Both exact-head reviews are now 5/5. The hosted native-server lane ran all 9 new ownership tests successfully. One unchanged tool-gateway rate-rule test failed in the first current-head CI run. It passed previously and passed locally, both narrowly and in its full 183-test file. The single same-head failed-job rerun passed. All current-head hosted gates are clean. The initial failure remains recorded; no test assertion was weakened. - The provider transport is an in-memory fixture. The managed wrapper, production checkpoint callback, control-plane result and Postgres ownership are real. No paid provider or live remote process is exercised. This does not prove a deployed recovery or a remote stop receipt. ## Risks - A slow required checkpoint extends the lifetime of the existing active renewal. Its cadence stays at 20 seconds for remote claims and 5 minutes for local claims. It runs only while the admitted checkpoint, close and collection remain active, then stops and joins on completion or unwinding. The tests observe a real remote renewal tick. No idle timer, new retry budget or new worker is added. - The release still checks company, task, run, owner, attempt, boot, PID, process start and expiry. Restart detachment retains its recovery owner. - The storage owner guard is unchanged. No migration, permission, provider setting or deployment change is included. ## Model Used OpenAI Codex, `gpt-6.1-sol`, with reasoning, source inspection, local code execution and read-only staging metadata. Context window and reasoning effort are host-managed. ## 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 - [x] All Paperclip CI gates are green - [x] 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
b8e8ab4641
commit
3a090b815b
6 files changed
+388
-99
No files matched your search
@@ -3832,6 +3832,18 @@ A successor claims its existing native coordinator and starts renewal before
|
||||
retiring a superseded warm session. An unclaimed preparation cannot authorize
|
||||
the predecessor's checkpoint writes. Shared attachment staging follows verified
|
||||
retirement; a failed claim or Stop cannot close that predecessor or stage files.
|
||||
|
||||
A completed native turn keeps its exact coordinator claim and the existing
|
||||
execution lease renewal until its required warm checkpoint, session close and
|
||||
instruction collection have joined. It then stops and joins renewal and releases
|
||||
only that claim. A checkpoint or collection failure keeps its initiating error
|
||||
while the same fenced release runs; a changed controller or owner is never
|
||||
cleared. Restart detachment retains its existing recovery owner and its typed
|
||||
outcome takes precedence over checkpoint, collection or renewal-stop errors.
|
||||
Only closed cleanup-failure flags accompany that outcome; raw secondary errors
|
||||
are not attached. Renewal exists
|
||||
only for this active obligation and stops on normal completion or unwinding; it
|
||||
does not create an idle worker or another recovery budget.
|
||||
Recovery also preserves the implicit isolated-home default when binding admitted
|
||||
configuration, so an unchanged home does not invalidate the retained session.
|
||||
Explicit home-mode changes retain their normal configuration-change checks.
|
||||
|
||||
@@ -322,6 +322,7 @@ const nativeRunnerSuitePath =
|
||||
"server/src/services/native-runtime/native-codex-runner.integration.test.ts";
|
||||
const ingressRetrySuitePath = "server/src/services/native-runtime/native-ingress-bootstrap-retry.integration.test.ts";
|
||||
const usageDeliverySuitePath = "server/src/services/native-runtime/native-usage-delivery.integration.test.ts";
|
||||
const warmCheckpointOwnershipSuitePath = "server/src/services/native-runtime/native-warm-checkpoint-ownership.integration.test.ts";
|
||||
const dotRunnerSuitePath = "server/src/__tests__/dot-runner.test.ts";
|
||||
const restartRecoverySuitePath =
|
||||
"server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts";
|
||||
@@ -369,7 +370,8 @@ test("12 PR without-chat shards plus the dedicated chat and native-runner lanes
|
||||
assert.ok(!files.includes(dotRunnerSuitePath));
|
||||
assert.ok(!files.includes(restartRecoverySuitePath));
|
||||
assert.ok(!files.includes(usageDeliverySuitePath));
|
||||
assert.deepEqual([...files, chatSuitePath, nativeRunnerSuitePath, ingressRetrySuitePath, dotRunnerSuitePath, restartRecoverySuitePath, ...restartWaitSuites, usageDeliverySuitePath, "server/src/services/native-runtime/plugin-runner-probe.test.ts"].sort(), full.selectedGeneralServerSuites.sort());
|
||||
assert.ok(!files.includes(warmCheckpointOwnershipSuitePath));
|
||||
assert.deepEqual([...files, chatSuitePath, nativeRunnerSuitePath, ingressRetrySuitePath, dotRunnerSuitePath, restartRecoverySuitePath, ...restartWaitSuites, usageDeliverySuitePath, warmCheckpointOwnershipSuitePath, "server/src/services/native-runtime/plugin-runner-probe.test.ts"].sort(), full.selectedGeneralServerSuites.sort());
|
||||
assert.equal(new Set(files).size, files.length);
|
||||
const defaultRun = dryRunJson([], prEnv);
|
||||
assert.ok(defaultRun.generalServerSuiteCount === full.generalServerSuiteCount);
|
||||
@@ -389,6 +391,7 @@ for (const [caller, envOverrides] of [["Release", { GITHUB_WORKFLOW: "Release" }
|
||||
assert.ok(files.includes(nativeRunnerSuitePath));
|
||||
assert.ok(files.includes(dotRunnerSuitePath));
|
||||
assert.ok(files.includes(restartRecoverySuitePath));
|
||||
assert.ok(files.includes(warmCheckpointOwnershipSuitePath));
|
||||
assert.deepEqual([...files, chatSuitePath].sort(), full.selectedGeneralServerSuites.sort());
|
||||
assert.equal(new Set(files).size, files.length);
|
||||
});
|
||||
@@ -421,6 +424,7 @@ test("the native-runner lane runs exactly the cargo-dependent suites", () => {
|
||||
restartRecoverySuitePath,
|
||||
...restartWaitSuites,
|
||||
usageDeliverySuitePath,
|
||||
warmCheckpointOwnershipSuitePath,
|
||||
]);
|
||||
});
|
||||
|
||||
|
||||
@@ -103,6 +103,7 @@ const nativeRunnerSuites = [
|
||||
"server/src/services/native-runtime/native-runner-restart-checkpoint-cost.integration.test.ts",
|
||||
"server/src/services/native-runtime/native-runner-restart-diskless.integration.test.ts",
|
||||
"server/src/services/native-runtime/native-usage-delivery.integration.test.ts",
|
||||
"server/src/services/native-runtime/native-warm-checkpoint-ownership.integration.test.ts",
|
||||
];
|
||||
// In the PR workflow (pr.yml, the caller of pr-trusted.yml — reusable
|
||||
// workflows inherit the caller's GITHUB_WORKFLOW), the last Verify Paperclip
|
||||
|
||||
+4
-3
@@ -392,17 +392,18 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", (
|
||||
onExecutionLeaseAcquired: () => preparation!.stop(),
|
||||
instructionWorkingCopy: {
|
||||
hasChanges: async () => {
|
||||
// Completion has released the coordinator, but the startup still
|
||||
// owns its session. A leftover preparation timer would detach it.
|
||||
const [settled] = await fixture.db.select().from(nativeRunFinalizations)
|
||||
.where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
expect(settled.leaseOwner).toBeNull();
|
||||
expect(settled).toMatchObject({ leaseOwner: claim.leaseOwner, attempt: claim.providerAttempt });
|
||||
await new Promise(resolve => setTimeout(resolve, 21_000));
|
||||
return false;
|
||||
},
|
||||
collectStopped: async () => undefined,
|
||||
},
|
||||
}).finally(() => preparation?.stop().catch(() => undefined));
|
||||
const [released] = await fixture.db.select().from(nativeRunFinalizations)
|
||||
.where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
expect(released.leaseOwner).toBeNull();
|
||||
expect(logs.filter(line => line.includes("controller authority detached for restart"))).toHaveLength(1);
|
||||
const recoveredCalls = await runner.execute({ command: "cat", args: [`${remoteCwd}/provider-calls.log`] });
|
||||
expect(recoveredCalls.stdout.split("\n").filter(line => line === "turn/start")).toHaveLength(1);
|
||||
|
||||
@@ -328,9 +328,12 @@ class NativePostResultFailureError extends Error {
|
||||
}
|
||||
|
||||
export class NativeControllerDetachedForRestartError extends Error {
|
||||
constructor() {
|
||||
readonly diagnostic?: Readonly<{ stage: "post_result_cleanup"; postExecutionCleanupFailed: boolean; renewalStopFailed: boolean }>;
|
||||
|
||||
constructor(diagnostic?: { stage: "post_result_cleanup"; postExecutionCleanupFailed: boolean; renewalStopFailed: boolean }) {
|
||||
super("native_controller_detached_for_restart");
|
||||
this.name = "NativeControllerDetachedForRestartError";
|
||||
if (diagnostic) this.diagnostic = Object.freeze(diagnostic);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -10506,7 +10509,6 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
expectedCurrentWakeComments,
|
||||
);
|
||||
}
|
||||
await leaseRenewal.stop();
|
||||
await trace.record({
|
||||
name: "native.result.finalize",
|
||||
parentName: "task.settle",
|
||||
@@ -11141,62 +11143,117 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
await timeController("github_stop", () => githubAccess!.stop());
|
||||
}
|
||||
}
|
||||
if (
|
||||
planSynchronizations.length === 0 &&
|
||||
"executionMode" in input.execution &&
|
||||
input.execution.executionMode === "plan"
|
||||
) {
|
||||
const markdown = semanticProviderPlanMarkdown(
|
||||
native.result as unknown as Record<string, unknown>,
|
||||
);
|
||||
if (markdown) {
|
||||
const digest = createHash("sha256").update(markdown).digest("hex");
|
||||
await recordPlanSynchronization({
|
||||
sourceEventId: `semantic-plan:${input.execution.binding.runId}:${digest}`,
|
||||
...(native.turnId ? { turnId: native.turnId } : {}),
|
||||
eventType: "plan.updated",
|
||||
payload: {
|
||||
schema: "paperclip.plan.updated.v1",
|
||||
planId: `semantic:${native.turnId ?? input.execution.binding.runId}`,
|
||||
revision: 1,
|
||||
complete: true,
|
||||
markdown,
|
||||
source: "semantic_result_artifact",
|
||||
},
|
||||
});
|
||||
let postExecutionFailed = false;
|
||||
let postExecutionCleanupFailed = false;
|
||||
try {
|
||||
if (warmSessionId !== null && lifecyclePolicy.mode === "warm") {
|
||||
const retainSession = native.terminal.runTerminalState === "succeeded" && native.result.continuation?.kind !== "case_handoff";
|
||||
const instructionCopy = input.instructionWorkingCopy;
|
||||
const ownedSession = warmNativeSessions.get(warmSessionId);
|
||||
let collectInstructions: boolean;
|
||||
try {
|
||||
collectInstructions = Boolean(instructionCopy && ownedSession?.ownerToken === warmSessionOwnerToken &&
|
||||
(instructionCopy.checkpointWarm ? !await timeController("instruction_checkpoint", () => instructionCopy.checkpointWarm!()) : await timeController("instruction_changes", () => instructionCopy.hasChanges())));
|
||||
} catch (error) {
|
||||
await releaseWarmNativeSession(warmSessionId, warmSessionOwnerToken, lifecyclePolicy.idleTimeoutMs, true)
|
||||
.catch(() => undefined);
|
||||
throw error;
|
||||
}
|
||||
const collectedByOwner = ownedSession?.instructionCopy?.collectStopped === instructionCopy?.collectStopped
|
||||
|| ownedSession?.instructionWorkingCopy?.collectStopped === instructionCopy?.collectStopped;
|
||||
if (collectInstructions && ownedSession) {
|
||||
ownedSession.closeOnReleaseReason = "registered instruction edits require stopped-provider collection";
|
||||
}
|
||||
await timeController("warm_release", () => releaseWarmNativeSession(
|
||||
warmSessionId,
|
||||
warmSessionOwnerToken,
|
||||
lifecyclePolicy.idleTimeoutMs,
|
||||
!retainSession,
|
||||
));
|
||||
if (collectInstructions && !collectedByOwner) await timeController("instruction_collection", () => instructionCopy!.collectStopped());
|
||||
}
|
||||
assertControllerActive();
|
||||
if (
|
||||
planSynchronizations.length === 0 &&
|
||||
"executionMode" in input.execution &&
|
||||
input.execution.executionMode === "plan"
|
||||
) {
|
||||
const markdown = semanticProviderPlanMarkdown(
|
||||
native.result as unknown as Record<string, unknown>,
|
||||
);
|
||||
if (markdown) {
|
||||
const digest = createHash("sha256").update(markdown).digest("hex");
|
||||
await recordPlanSynchronization({
|
||||
sourceEventId: `semantic-plan:${input.execution.binding.runId}:${digest}`,
|
||||
...(native.turnId ? { turnId: native.turnId } : {}),
|
||||
eventType: "plan.updated",
|
||||
payload: {
|
||||
schema: "paperclip.plan.updated.v1",
|
||||
planId: `semantic:${native.turnId ?? input.execution.binding.runId}`,
|
||||
revision: 1,
|
||||
complete: true,
|
||||
markdown,
|
||||
source: "semantic_result_artifact",
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
postExecutionFailed = true;
|
||||
postExecutionCleanupFailed = !(error instanceof NativeControllerDetachedForRestartError);
|
||||
throw error;
|
||||
} finally {
|
||||
let renewalStopFailed = false;
|
||||
try {
|
||||
try {
|
||||
await leaseRenewal.stop();
|
||||
} catch (error) {
|
||||
renewalStopFailed = true;
|
||||
throw error;
|
||||
}
|
||||
if (nativeRunsDetachingForRestart.has(input.execution.binding.runId)) {
|
||||
throw new NativeControllerDetachedForRestartError();
|
||||
}
|
||||
const releaseNow = new Date();
|
||||
const released = await timeController("coordinator_release", () => input.db.transaction(async tx => {
|
||||
await notifyNativeFinalizationWork(tx, input.execution.binding);
|
||||
return tx.update(nativeRunFinalizations)
|
||||
.set({
|
||||
leaseOwner: null,
|
||||
leaseExpiresAt: null,
|
||||
recoveryState: null,
|
||||
recoveryRequestId: null,
|
||||
updatedAt: releaseNow,
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(nativeRunFinalizations.runId, input.execution.binding.runId),
|
||||
eq(nativeRunFinalizations.companyId, input.execution.binding.companyId),
|
||||
eq(nativeRunFinalizations.issueId, input.execution.binding.issueId),
|
||||
eq(nativeRunFinalizations.leaseOwner, leaseOwner),
|
||||
eq(nativeRunFinalizations.attempt, attempt),
|
||||
eq(nativeRunFinalizations.controllerBootId, controller.bootId),
|
||||
eq(nativeRunFinalizations.controllerPid, controller.pid),
|
||||
eq(
|
||||
nativeRunFinalizations.controllerProcessStartedAt,
|
||||
controller.processStartedAt,
|
||||
),
|
||||
gt(nativeRunFinalizations.leaseExpiresAt, sql`now()`),
|
||||
),
|
||||
)
|
||||
.returning({ runId: nativeRunFinalizations.runId })
|
||||
.then((rows) => rows[0] ?? null);
|
||||
}));
|
||||
if (!released) throw new Error("native_session_lease_lost");
|
||||
} catch (error) {
|
||||
if (nativeRunsDetachingForRestart.has(input.execution.binding.runId)) {
|
||||
throw new NativeControllerDetachedForRestartError({ stage: "post_result_cleanup",
|
||||
postExecutionCleanupFailed, renewalStopFailed });
|
||||
}
|
||||
if (!postExecutionFailed) throw error;
|
||||
logger.warn({ runId: input.execution.binding.runId }, "Native finalization claim release failed after an earlier error");
|
||||
}
|
||||
}
|
||||
const releaseNow = new Date();
|
||||
const released = await timeController("coordinator_release", () => input.db.transaction(async tx => {
|
||||
await notifyNativeFinalizationWork(tx, input.execution.binding);
|
||||
return tx.update(nativeRunFinalizations)
|
||||
.set({
|
||||
leaseOwner: null,
|
||||
leaseExpiresAt: null,
|
||||
recoveryState: null,
|
||||
recoveryRequestId: null,
|
||||
updatedAt: releaseNow,
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(nativeRunFinalizations.runId, input.execution.binding.runId),
|
||||
eq(nativeRunFinalizations.companyId, input.execution.binding.companyId),
|
||||
eq(nativeRunFinalizations.issueId, input.execution.binding.issueId),
|
||||
eq(nativeRunFinalizations.leaseOwner, leaseOwner),
|
||||
eq(nativeRunFinalizations.attempt, attempt),
|
||||
eq(nativeRunFinalizations.controllerBootId, controller.bootId),
|
||||
eq(nativeRunFinalizations.controllerPid, controller.pid),
|
||||
eq(
|
||||
nativeRunFinalizations.controllerProcessStartedAt,
|
||||
controller.processStartedAt,
|
||||
),
|
||||
gt(nativeRunFinalizations.leaseExpiresAt, sql`now()`),
|
||||
),
|
||||
)
|
||||
.returning({ runId: nativeRunFinalizations.runId })
|
||||
.then((rows) => rows[0] ?? null);
|
||||
}));
|
||||
if (!released) throw new Error("native_session_lease_lost");
|
||||
const finalization: NativeFinalizationResult = {
|
||||
schema: "paperclip.native-finalization.v1",
|
||||
runtimeMode: "native",
|
||||
@@ -11215,45 +11272,6 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
highestContiguousSourceSeq: native.highestContiguousSourceSeq,
|
||||
workspaceFinalizeStatus: "pending",
|
||||
};
|
||||
// A following run cannot attach until the prior run's durable finalization
|
||||
// is committed. Provider completion alone is not an authority boundary.
|
||||
if (warmSessionId !== null && lifecyclePolicy.mode === "warm") {
|
||||
// A structured failed/cancelled result completes the protocol without
|
||||
// throwing. Heartbeat still stops its sandbox, so retire the provider now
|
||||
// while its transport can collect files and checkpoint the suspended runner.
|
||||
// Leaving it in the warm map makes the next turn retire a stopped transport.
|
||||
// A different agent may next use this same task workspace. Persist this
|
||||
// agent's session checkpoint, then close its provider process before the
|
||||
// heartbeat release proof. Returning later resumes its own saved session.
|
||||
const retainSession = native.terminal.runTerminalState === "succeeded" && native.result.continuation?.kind !== "case_handoff";
|
||||
const instructionCopy = input.instructionWorkingCopy;
|
||||
const ownedSession = warmNativeSessions.get(warmSessionId);
|
||||
let collectInstructions: boolean;
|
||||
try {
|
||||
collectInstructions = Boolean(instructionCopy && ownedSession?.ownerToken === warmSessionOwnerToken &&
|
||||
(instructionCopy.checkpointWarm ? !await timeController("instruction_checkpoint", () => instructionCopy.checkpointWarm!()) : await timeController("instruction_changes", () => instructionCopy.hasChanges())));
|
||||
} catch (error) {
|
||||
// A checkpoint or its receipt callback can reject. Retire this owner
|
||||
// before heartbeat stops the sandbox, preserving the initiating error.
|
||||
await releaseWarmNativeSession(warmSessionId, warmSessionOwnerToken, lifecyclePolicy.idleTimeoutMs, true)
|
||||
.catch(() => undefined);
|
||||
throw error;
|
||||
}
|
||||
const collectedByOwner = ownedSession?.instructionCopy?.collectStopped === instructionCopy?.collectStopped
|
||||
|| ownedSession?.instructionWorkingCopy?.collectStopped === instructionCopy?.collectStopped;
|
||||
if (collectInstructions && ownedSession) {
|
||||
// Keep the unchanged warm path intact. A changed private instruction copy
|
||||
// requires the existing checkpoint-and-close boundary before collection.
|
||||
ownedSession.closeOnReleaseReason = "registered instruction edits require stopped-provider collection";
|
||||
}
|
||||
await timeController("warm_release", () => releaseWarmNativeSession(
|
||||
warmSessionId,
|
||||
warmSessionOwnerToken,
|
||||
lifecyclePolicy.idleTimeoutMs,
|
||||
!retainSession,
|
||||
));
|
||||
if (collectInstructions && !collectedByOwner) await timeController("instruction_collection", () => instructionCopy!.collectStopped());
|
||||
}
|
||||
const providerFailure = providerFailureObservation.forTerminal(native.turnId, native.terminal);
|
||||
const providerExplanation = providerFailureObservation.providerFailureForTerminal(native.turnId, native.terminal);
|
||||
// Keep the generic failure behavior, but record why the turn failed.
|
||||
|
||||
+253
@@ -0,0 +1,253 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { mkdtemp, rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, expect, it, vi } from "vitest";
|
||||
import { completionContracts, heartbeatRuns, issues, nativeRunFinalizations } from "@paperclipai/db";
|
||||
import type { NativeExecutionInputV1 } from "@paperclipai/paperclip-runner";
|
||||
import { createLocalDiskStorageProvider } from "../../storage/local-disk-provider.js";
|
||||
import { createStorageService } from "../../storage/service.js";
|
||||
import { describeEmbeddedPostgres, nativeRestartFixture } from "../../__tests__/helpers/native-restart.js";
|
||||
import { CONTROL_PLANE_CONFORMANCE_RESULT } from "../../vendor/paperclip-runner/testing.js";
|
||||
import { defaultCapabilityRunnerdBinary } from "../../vendor/paperclip-runner/index.js";
|
||||
|
||||
type ProviderTransport = ReturnType<NonNullable<NonNullable<Parameters<typeof import("../../vendor/paperclip-runner/index.js").createNativeSessionBackend>[1]>["codexTransportFactory"]>>;
|
||||
const { FakeCodexTransport } = await import(new URL("../../../../packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.test-support.ts", import.meta.url).href) as { FakeCodexTransport: new () => ProviderTransport };
|
||||
const { runnerdRecoveryInternals } = await import(new URL("../../../../packages/paperclip-runner/src/live/runnerd-codex-transport.ts", import.meta.url).href) as {
|
||||
runnerdRecoveryInternals: { releaseRunnerProcessOwnership(input: { runnerSettled: boolean; checkpoint: typeof state.checkpoint;
|
||||
release: () => Promise<void>; forceKill: () => Promise<void> }): Promise<void> };
|
||||
};
|
||||
|
||||
const state = vi.hoisted(() => ({ execute: vi.fn(), checkpoint: null as null | ((settlement: "settled" | "unsettled") => Promise<void>), storage: null as unknown, transport: null as unknown }));
|
||||
vi.mock("./native-harness-backup-storage.js", async original => ({
|
||||
...await original<typeof import("./native-harness-backup-storage.js")>(),
|
||||
durableNativeHarnessStorage: () => state.storage,
|
||||
}));
|
||||
vi.mock("./native-runner-execution-driver.js", async original => {
|
||||
const actual = await original<typeof import("./native-runner-execution-driver.js")>();
|
||||
return { ...actual, createNativeRunnerExecutionDriver: (input: Parameters<typeof actual.createNativeRunnerExecutionDriver>[0]) => {
|
||||
if (input.checkpoint) state.checkpoint = input.checkpoint as typeof state.checkpoint;
|
||||
return actual.createNativeRunnerExecutionDriver(input);
|
||||
} };
|
||||
});
|
||||
vi.mock("../../vendor/paperclip-runner/index.js", async original => {
|
||||
const actual = await original<typeof import("../../vendor/paperclip-runner/index.js")>();
|
||||
return { ...actual, executeNativeSession: state.execute,
|
||||
createNativeSessionBackend: (input: Parameters<typeof actual.createNativeSessionBackend>[0], options: Parameters<typeof actual.createNativeSessionBackend>[1]) => {
|
||||
options?.codexTransportFactory?.();
|
||||
return actual.createNativeSessionBackend(input, { ...options, codexTransportFactory: () => state.transport as ProviderTransport });
|
||||
} };
|
||||
});
|
||||
import { closeWarmNativeSessionsForEnvironment, detachNativeSessionsForRestart, executePaperclipNativeSession, NativeControllerDetachedForRestartError } from "./native-session-executor.js";
|
||||
import * as executionLease from "./native-execution-lease.js";
|
||||
import { finalizeNativeRun } from "./native-run-finalizer.js";
|
||||
import { prepareNativeHeartbeatRun } from "./prepare-native-run.js";
|
||||
import { createPostgresWakeQueueAdapter } from "../../modules/wake-queue/adapters/postgres.js";
|
||||
import { createReleaseIssueExecution } from "../../modules/wake-queue/application/use-cases.js";
|
||||
|
||||
describeEmbeddedPostgres("managed warm checkpoint ownership", () => {
|
||||
const environment = nativeRestartFixture();
|
||||
let root: string;
|
||||
beforeAll(async () => { root = await mkdtemp(join(tmpdir(), "warm-checkpoint-owner-")); });
|
||||
afterAll(async () => { state.storage = null; await rm(root, { recursive: true, force: true }); });
|
||||
|
||||
it.each(["normal", "delayed", "instruction_error", "fresh_successor", "wrong_boot", "superseded_owner", "detach_during_close", "detach_during_instruction_error", "detach_during_instruction_error_and_renewal_loss"])("keeps the producing coordinator claim until the managed failed-turn close joins: %s", async scenario => {
|
||||
let fixture: Omit<Awaited<ReturnType<typeof environment.seedRun>>, "runId"> & { runId: string } = await environment.seedRun(`WARM-CHECKPOINT-OWNER-${scenario}`);
|
||||
const { companyId, agentId } = environment.current;
|
||||
const [contract] = await fixture.db.select().from(completionContracts).where(eq(completionContracts.issueId, fixture.issueId));
|
||||
let execution: NativeExecutionInputV1 = {
|
||||
schema: "paperclip.native-execution-input.v1",
|
||||
binding: { companyId, agentId, issueId: fixture.issueId, runId: fixture.runId, executionWorkspaceId: randomUUID() },
|
||||
provider: { kind: "codex", model: "test-model" },
|
||||
task: { identifier: "WARM-1", title: "Retain checkpoint owner", description: null, prompt: "Return a failed turn.", workMode: "standard" },
|
||||
workspace: { cwd: root, repoUrl: null, repoRef: null, branchName: null },
|
||||
session: { normalizedSessionId: fixture.native.normalizedSessionId, driverKind: "codex_app_server", protocolVersion: 1,
|
||||
lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 } },
|
||||
completionContract: { id: contract.id, sha256: contract.canonicalSha256, schemaVersion: "paperclip.completion-contract.v1",
|
||||
contract: { ...contract.contractJson, revision: String(contract.revision) } } as NativeExecutionInputV1["completionContract"],
|
||||
interactionResponses: [], credentialBindings: [],
|
||||
};
|
||||
state.storage = createStorageService(createLocalDiskStorageProvider(join(root, "objects")));
|
||||
const transport = new FakeCodexTransport();
|
||||
const request = transport.request.bind(transport);
|
||||
transport.request = async (method, params) => {
|
||||
const response = await request(method, params);
|
||||
return method === "thread/start" || method === "thread/resume"
|
||||
? { ...response, cwd: root, thread: { ...response.thread as Record<string, unknown>, cwd: root } }
|
||||
: response;
|
||||
};
|
||||
state.transport = transport;
|
||||
const closeFailures: string[] = [];
|
||||
const ownersAtClose: Array<{ phase: string; leaseOwner: string | null; resultId: string | null; leaseExpiresAt: Date | null }> = [];
|
||||
const initiatingError = new Error("fixture instruction checkpoint failed private-canary-N45");
|
||||
const invalidated = scenario === "wrong_boot" || scenario === "superseded_owner";
|
||||
const renewalLoss = scenario === "detach_during_instruction_error_and_renewal_loss";
|
||||
const instructionErrorWithDetach = scenario.startsWith("detach_during_instruction_error");
|
||||
let originalBoot: string | null = null;
|
||||
let replacementOwner: string | null = null;
|
||||
let enteredClose!: () => void, releaseClose!: () => void;
|
||||
const closeEntered = new Promise<void>(resolve => { enteredClose = resolve; });
|
||||
const closeGate = new Promise<void>(resolve => { releaseClose = resolve; });
|
||||
const detached = vi.fn(async () => {});
|
||||
Object.assign(transport, { detachControllerForRestart: detached });
|
||||
const renew = executionLease.startNativeSessionExecutionLeaseRenewal;
|
||||
let renewalStopped = false;
|
||||
const renewal = vi.spyOn(executionLease, "startNativeSessionExecutionLeaseRenewal").mockImplementation(input => {
|
||||
const owned = renew(input);
|
||||
return { ...owned, stop: async () => { renewalStopped = true; await owned.stop(); } };
|
||||
});
|
||||
transport.close = async () => {
|
||||
const [owner] = await fixture.db.select({ phase: nativeRunFinalizations.phase, leaseOwner: nativeRunFinalizations.leaseOwner,
|
||||
resultId: nativeRunFinalizations.resultId, leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt }).from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
ownersAtClose.push(owner);
|
||||
expect(renewalStopped).toBe(false);
|
||||
if (scenario === "detach_during_close") { enteredClose(); await closeGate; }
|
||||
if (scenario === "delayed") {
|
||||
await expect.poll(async () => {
|
||||
const [current] = await fixture.db.select({ expires: nativeRunFinalizations.leaseExpiresAt }).from(nativeRunFinalizations)
|
||||
.where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
return current.expires!.getTime();
|
||||
}, { timeout: 25_000, interval: 250 }).toBeGreaterThan(owner.leaseExpiresAt!.getTime() + 10_000);
|
||||
expect(renewalStopped).toBe(false);
|
||||
}
|
||||
try {
|
||||
await runnerdRecoveryInternals.releaseRunnerProcessOwnership({ runnerSettled: false,
|
||||
checkpoint: state.checkpoint, release: async () => {}, forceKill: async () => {} });
|
||||
} catch (error) { closeFailures.push(String(error)); throw error; }
|
||||
};
|
||||
state.execute.mockReset().mockImplementation(async options => {
|
||||
const session = await options.backend.openSession({ identity: { companyId, agentId, issueId: fixture.issueId,
|
||||
runId: fixture.runId, sessionId: execution.session.normalizedSessionId }, workingDirectory: root });
|
||||
await options.onSession(session);
|
||||
await options.controlPlane.openRun({ identity: { companyId, agentId, issueId: fixture.issueId, runId: fixture.runId,
|
||||
sessionId: execution.session.normalizedSessionId }, backendKind: "mock", sourceInstanceId: fixture.native.runnerInstanceId });
|
||||
await options.controlPlane.appendEvent({ schema: "paperclip.prp.event.v1", runId: fixture.runId,
|
||||
normalizedSessionId: execution.session.normalizedSessionId, turnId: fixture.native.turnId,
|
||||
sourceInstanceId: fixture.native.runnerInstanceId, sourceEventId: "provider-overload", sourceSeq: 1,
|
||||
sourceKind: "runner", eventType: "turn.failed", schemaVersion: 1, priority: 0, emittedAt: new Date().toISOString(),
|
||||
payload: { status: "failed", error: { codexErrorInfo: "serverOverloaded" } } });
|
||||
const terminal = { schema: "paperclip.prp.terminal.v1", runTerminalState: "failed", turnTerminalState: "failed", reportedWorkDisposition: "needs_review" };
|
||||
const result = { ...CONTROL_PLANE_CONFORMANCE_RESULT, reportedWorkDisposition: "needs_review",
|
||||
completionClaim: { ...CONTROL_PLANE_CONFORMANCE_RESULT.completionClaim, contractRevision: execution.completionContract.contract.revision } };
|
||||
await options.controlPlane.completeRun({ result, terminal, turnId: fixture.native.turnId });
|
||||
return { result, terminal, turnId: fixture.native.turnId, normalizedSessionId: execution.session.normalizedSessionId,
|
||||
providerSessionId: "fixture-session", driverKind: "codex_app_server", driverVersion: "fixture", nativeEventCount: 1, highestContiguousSourceSeq: 1, usage: null };
|
||||
});
|
||||
const environmentId = randomUUID();
|
||||
let acquisition: "created" | "resumed" = "created";
|
||||
const providerLeaseId = randomUUID();
|
||||
const runCurrent = async () => {
|
||||
await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: { ...fixture.preparedProfile, nativeExecutionInput: execution } })
|
||||
.where(eq(heartbeatRuns.id, fixture.runId));
|
||||
return executePaperclipNativeSession({ db: fixture.db, execution, runnerInstanceId: fixture.native.runnerInstanceId,
|
||||
useRunnerd: true, runnerIngressAuthorized: true,
|
||||
runnerRemoteBinaryPath: defaultCapabilityRunnerdBinary(), runnerRemoteCodexPath: defaultCapabilityRunnerdBinary(),
|
||||
runnerExecutionTarget: { kind: "remote", transport: "sandbox", remoteCwd: root, environmentId,
|
||||
leaseId: randomUUID(), providerKey: "fixture", effectiveCapabilities: { runnerWebSocketIngress: true },
|
||||
sandboxLeaseAcquisition: { outcome: acquisition, providerLeaseId },
|
||||
runner: { execute: async (command: { args?: string[] }) => command.args?.[1] === "uname -s; uname -m"
|
||||
? { exitCode: 0, timedOut: false, stdout: `${process.platform === "darwin" ? "Darwin" : "Linux"}\n${process.arch === "arm64" ? "arm64" : "x86_64"}\n`, stderr: "" }
|
||||
: { exitCode: 1, timedOut: false, stdout: "", stderr: "" }, syncIn: async () => {} } } as never,
|
||||
instructionWorkingCopy: { runId: fixture.runId, root, checkpointWarm: async () => {
|
||||
if (instructionErrorWithDetach) { enteredClose(); await closeGate; throw initiatingError; }
|
||||
if (scenario === "instruction_error") throw initiatingError;
|
||||
if (invalidated) {
|
||||
const [current] = await fixture.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
originalBoot = current.controllerBootId;
|
||||
replacementOwner = scenario === "superseded_owner" ? randomUUID() : current.leaseOwner;
|
||||
await fixture.db.update(nativeRunFinalizations).set({ controllerBootId: randomUUID(), leaseOwner: replacementOwner,
|
||||
controllerGeneration: current.controllerGeneration + 1 }).where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
}
|
||||
return true;
|
||||
},
|
||||
collectStopped: async () => {}, hasChanges: async () => false },
|
||||
});
|
||||
};
|
||||
let running: ReturnType<typeof runCurrent> | undefined;
|
||||
try {
|
||||
running = runCurrent();
|
||||
if (scenario.startsWith("detach_during_")) {
|
||||
await Promise.race([closeEntered, running.then(() => { throw new Error("fixture checkpoint was not reached"); })]);
|
||||
expect(renewalStopped).toBe(false);
|
||||
const [beforeDetach] = await fixture.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
const detach = await detachNativeSessionsForRestart([fixture.runId]);
|
||||
expect(detach).toEqual({ detachedRunIds: [fixture.runId], inactiveRunIds: [], unsupportedRunIds: [], pendingRunIds: [] });
|
||||
expect(detached).toHaveBeenCalledOnce();
|
||||
const [held] = await fixture.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
expect(held).toMatchObject({ phase: "workspace_finalizing", leaseOwner: beforeDetach.leaseOwner, resultId: beforeDetach.resultId });
|
||||
if (renewalLoss) {
|
||||
originalBoot = held.controllerBootId;
|
||||
replacementOwner = randomUUID();
|
||||
await fixture.db.update(nativeRunFinalizations).set({ leaseOwner: replacementOwner,
|
||||
controllerGeneration: held.controllerGeneration + 1 }).where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
await expect.poll(() => renewal.mock.results[0].value.remainingMs(), { timeout: 25_000, interval: 250 }).toBe(0);
|
||||
}
|
||||
releaseClose();
|
||||
await expect(running).rejects.toBeInstanceOf(NativeControllerDetachedForRestartError);
|
||||
const failure = await running.catch(error => error);
|
||||
expect(failure).toMatchObject({ diagnostic: { stage: "post_result_cleanup",
|
||||
postExecutionCleanupFailed: instructionErrorWithDetach, renewalStopFailed: renewalLoss } });
|
||||
expect(failure).not.toHaveProperty("cause");
|
||||
expect(JSON.stringify(failure)).not.toContain("private-canary-N45");
|
||||
} else if (invalidated) await expect(running).rejects.toThrow("native_session_lease_lost");
|
||||
else if (scenario === "instruction_error") await expect(running).rejects.toBe(initiatingError);
|
||||
else {
|
||||
const result = await running;
|
||||
expect(result.exitCode).toBe(1);
|
||||
expect(result.errorCode).toBe("native_provider_overloaded");
|
||||
}
|
||||
expect(state.checkpoint).toBeTypeOf("function");
|
||||
if (invalidated) expect(closeFailures).toEqual(["Error: native_harness_storage_owner_missing"]);
|
||||
else expect(closeFailures).toEqual([]);
|
||||
expect(ownersAtClose).toHaveLength(1);
|
||||
expect(ownersAtClose[0].leaseOwner).not.toBeNull();
|
||||
expect(renewalStopped).toBe(true);
|
||||
const [released] = await fixture.db.select({ leaseOwner: nativeRunFinalizations.leaseOwner }).from(nativeRunFinalizations)
|
||||
.where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
expect(released.leaseOwner).toBe(invalidated || renewalLoss ? replacementOwner : scenario.startsWith("detach_during_") ? ownersAtClose[0].leaseOwner : null);
|
||||
expect(state.execute).toHaveBeenCalledTimes(1);
|
||||
if (scenario.startsWith("detach_during_")) {
|
||||
const rows = await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.nativeIssueId, fixture.issueId));
|
||||
expect(rows).toHaveLength(1);
|
||||
expect(rows[0].resultJson).not.toHaveProperty("nativeCancellation");
|
||||
}
|
||||
if (scenario === "fresh_successor") {
|
||||
await fixture.db.update(heartbeatRuns).set({ responsibleUserId: "local-capacity-owner" }).where(eq(heartbeatRuns.id, fixture.runId));
|
||||
await fixture.db.update(issues).set({ responsibleUserId: "local-capacity-owner" }).where(eq(issues.id, fixture.issueId));
|
||||
await finalizeNativeRun({ db: fixture.db, runId: fixture.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true });
|
||||
const [retry] = await fixture.db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.retryOfRunId, fixture.runId)));
|
||||
expect(retry).toMatchObject({ status: "scheduled_retry", scheduledRetryReason: "native_provider_overloaded", contextSnapshot: { forceFreshSession: true } });
|
||||
const unexpected = async () => { throw new Error("unexpected fixture escalation"); };
|
||||
const release = createReleaseIssueExecution({ issueLock: createPostgresWakeQueueAdapter(fixture.db, {
|
||||
resolveResponsibleUserId: unexpected, getRoutineEnv: unexpected, resolveSessionBeforeForWakeup: unexpected }),
|
||||
recovery: { escalateStrandedAssignedIssue: unexpected, escalateStrandedRecoveryIssueInPlace: unexpected } });
|
||||
expect((await release({ companyId, runId: fixture.runId, now: new Date() })).outcome.kind).toBe("released");
|
||||
const [ready] = await fixture.db.update(heartbeatRuns).set({ status: "running", runtimeMode: "native", nativeIssueId: fixture.issueId })
|
||||
.where(eq(heartbeatRuns.id, retry.id)).returning();
|
||||
await fixture.db.update(issues).set({ executionRunId: retry.id }).where(eq(issues.id, fixture.issueId));
|
||||
const [task] = await fixture.db.select().from(issues).where(eq(issues.id, fixture.issueId));
|
||||
const native = await prepareNativeHeartbeatRun({ db: fixture.db, run: ready, issue: task, environmentLeaseId: randomUUID() });
|
||||
await fixture.db.insert(nativeRunFinalizations).values({ runId: retry.id, companyId, issueId: fixture.issueId, phase: "observed" });
|
||||
const previousSession = execution.session.normalizedSessionId;
|
||||
execution = { ...execution, binding: { ...execution.binding, runId: retry.id }, session: { ...execution.session, normalizedSessionId: native.normalizedSessionId } };
|
||||
expect(execution.session.normalizedSessionId).not.toBe(previousSession);
|
||||
fixture = { ...fixture, runId: retry.id, native };
|
||||
acquisition = "resumed";
|
||||
renewalStopped = false;
|
||||
const second = await runCurrent();
|
||||
expect(second.errorCode).toBe("native_provider_overloaded");
|
||||
expect(state.execute).toHaveBeenCalledTimes(2);
|
||||
expect(ownersAtClose).toHaveLength(2);
|
||||
expect(closeFailures).toEqual([]);
|
||||
}
|
||||
} finally {
|
||||
releaseClose();
|
||||
await running?.catch(() => undefined);
|
||||
if (invalidated || renewalLoss) await fixture.db.update(nativeRunFinalizations).set({ controllerBootId: originalBoot, phase: "committed",
|
||||
leaseOwner: null, leaseExpiresAt: null }).where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
await closeWarmNativeSessionsForEnvironment({ environmentId, reason: "fixture cleanup" });
|
||||
renewal.mockRestore();
|
||||
}
|
||||
}, 45_000);
|
||||
});
|
||||
Reference in new issue
Block a user