RomanNum3ral
/
paperclip
mirror of https://github.com/paperclipai/paperclip.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix: continue conversations after confirmed remote runner stop (#13254) 

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Users can stop a run and send another message on the same task.
> - Remote runners need evidence from their sandbox provider that
execution stopped.
> - Local process checks cannot prove that a remote process exited.
> - This pull request records provider stop receipts and uses them for
conversation admission.
> - New user messages can proceed after confirmed cleanup without
repeating interrupted actions.

## Linked Issues or Issue Description

Refs #13237 and #13239. Related: #13163 covers app-restart recovery;
this change covers an explicit stop followed by a new user message.

**What happened?**

A stopped remote Claude ACP task kept its execution hold after Daytona
cleanup succeeded. Native runners also rejected remote process
identities and retained stale session cleanup gates. A message sent
during cleanup could stay deferred after the sandbox stopped.

**Expected behavior**

After the provider confirms that the old execution stopped, a new user
message starts a fresh turn. Pending user messages must not need another
message to trigger admission. Prior action outcomes remain recorded.

**Steps to reproduce**

1. Start a long-running task in Daytona with a legacy Claude ACP or
native ACP runner.
2. Cancel the run while its tool is active.
3. Send a new message immediately, or after cleanup completes.
4. Observe the execution hold despite the old sandbox having stopped.

**Paperclip version or commit**

Reproduced on master at 7b829efdf6. The
branch is rebased on current master.

## What Changed

- Add optional provider stop receipts to sandbox release and destroy
hooks. Old plugins remain compatible.
- Bind receipts to the company, run, lease, and provider resource.
Failed cleanup cannot supply stop authority.
- Acknowledge legacy remote cancellation after confirmed termination.
- Admit native user continuations using remote receipts instead of host
process checks.
- Retire only the settled cleanup owner matching the stopped company,
run, and provider resource. Isolate cleanup gates between remote
sandboxes, including two sandboxes owned by one run.
- Reconsider user messages deferred during remote cleanup through normal
admission, including successful later cleanup retries.
- Preserve receipts through cleanup retries and inline cleanup after
failed startup.
- Permit provider destruction after a terminal remote checkpoint
failure. Busy ownership still blocks destruction.
- Add regression tests and update execution semantics.
- Give the real preview fixture ten seconds for cold startup. Reuse
release-mode Rust artifacts for the filtered parity checks, avoiding a
duplicate debug test build that exhausted CI disk twice; test filters
and assertions are unchanged.

## Verification

- Recursive typecheck and build passed.
- Targeted server, Daytona plugin, and native runtime tests passed. They
cover missing or mismatched receipts, failed cleanup, local process
protection, exact cleanup ownership, and a message sent during cleanup.
- Live Daytona tests passed for legacy Claude ACP, native per-turn,
native warm, and a newly created native runner using disposable
sandboxes. Each original run was cancelled; its explicit follow-up
completed with no execution hold. The disposable case confirmed a new
sandbox after deletion.
- Native tests used the provider's Opus 5 selector, `opus[1m]`, and the
Linux runner bundle from the sandbox image. Existing checkpoint/sync
finalization warnings remained visible before the native follow-ups
reached committed success. This change does not repair those separate
warnings or guarantee recovery of uncheckpointed files.
- A direct live provider test also passed with the final delete-wait
change: the destroy hook returned its receipt only after Daytona
reported the sandbox destroyed. The final Daytona plugin suite passed
all 153 tests; its build passed.
- After rebasing on master, 284 targeted server/plugin tests and 313
native executor tests passed. The native session runtime suite passed
all 128 tests.
- The review regression passed all 145 tests across the continuation,
environment runtime, and pending-cleanup sweep suites. Recursive
typecheck and build passed again after that fix.
- The security review's exact-resource finding is fixed. Cleanup
completion requires the same provider-resource scope used during session
creation. All 128 native runtime tests, 61 server continuation/cleanup
tests, recursive typecheck, and build passed after this fix. The
two-sandboxes-in-one-run regression proves one receipt cannot retire the
other quarantine.
- Greptile reviewed the final commit at 5/5, the security scan passed,
and no review threads remain unresolved. The final native executor suite
also passed all 313 tests.
- The full local suite reached 10,616 passing tests before stopping on
four failures. All four now pass in focused reruns: the final-code
continuation/teardown suites, the attachment suite, and the native
session test after building its required fake-provider binary. The
original full run overlapped source edits and did not reach the
remaining groups; it is not counted as a full-suite pass.
- The preview exposure suite passed all 25 applicable tests (three
Linux-only tests skipped locally) after the startup allowance change.
Both release-mode Rust parity commands passed locally. All final-head PR
checks passed, including full server/workspace/browser test groups, full
native runner verification, build, typecheck, canary dry-run, and the
aggregate gates.

## Risks

- A provider must return a receipt only after confirmed termination.
Incorrect provider claims could permit overlapping execution.
- Older providers without receipts retain the existing hold. Missing
evidence, failed cleanup, active ownership, pauses, approvals, and
budgets still block admission.
- This change preserves unknown action outcomes and old checkpoints. It
does not authorize replay or alter historical runs.
- No database migration or telemetry contract change is required.

## Model Used

OpenAI GPT-6 through Codex, with repository inspection, code execution,
and browser tools. The exact backend revision and context-window size
are not exposed in this session.

## 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 references)
- [x] My branch name describes the change and contains no internal
Paperclip ticket id or instance-derived details
- [x] I have run targeted 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:
Dotta 2026-09-11 15:18:37 -05:00 committed by GitHub
parent a23ae894a5
commit 663c44cb2b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
21 changed files with 534 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
doc/execution-semantics.md
View File

@ -837,7 +837,7 @@ Shutdown, process loss, and provider failure use the existing durable failure re
Real gates still apply: company and task ownership, active provider ownership, budget limits, agent availability, dependencies, pending approval/review paths, and explicit pause holds. Native runner reattachment and finalization retain their existing ownership protocol. Process, HTTP, and gateway adapters retain their recovery rules because invoking those adapters can itself repeat an external action rather than start a conversation turn.
An operator Stop still waits for local provider termination. Stop alone never promotes deferred comments or starts an automatic continuation. Once stopped, the next explicit wake adopts pending comment IDs in order through the existing queue. A compatible saved ACP session can resume, and an unavailable or incompatible session can start fresh with the full task context. Run credentials and scratch paths remain scoped to the new run. A subtree pause requires Resume; a message does not bypass it.
An operator Stop waits for provider termination. Remote sandbox providers may return a stopped/deleted receipt after their control-plane operation completes. Paperclip binds that receipt to the company, run, and exact lease; successful file cleanup, a terminal run row, or an in-sandbox shutdown event is not sufficient. Legacy conversational runs receive their cancellation acknowledgement after all remote leases have confirmed termination. Stop alone never creates a continuation. A user message queued during remote cleanup is reconsidered when the provider confirms termination; it still passes normal admission and adopts pending comment IDs in order. Once stopped, the next explicit wake uses the same queue. A compatible saved ACP session can resume, and an unavailable or incompatible session can start fresh with the full task context. Run credentials and scratch paths remain scoped to the new run. A subtree pause requires Resume; a message does not bypass it.
Historical legacy interruption holds for conversational adapters no longer block new messages or Resume. Classification uses the run’s saved adapter invocation or continuation policy, never the agent’s current adapter settings. Missing historical adapter evidence retains the hold. A terminal row with a live predecessor process or unreleased environment lease still blocks actual admission and Resume. Retry scheduling can happen before cleanup, but grants no execution authority. Recovery folds their obsolete no-replay bookkeeping without changing task ownership, status, or automatically waking old work. The audit trail remains readable. Native integrity and ownership holds, and non-conversational adapter holds, remain enforced.
@ -884,9 +884,13 @@ request, task history, completed work, and the interruption notice. It receives
no instruction to repeat old tool calls. Later messages cannot reset the old
incident's retry budget or create another automatic replacement for it.
The initial native admission path verifies local process identities. Missing
process identity or remote ownership without a target-aware stop proof remains a
hold; a terminal database status or a PID check on the wrong host is insufficient.
Native admission verifies local process identities for local runs. Remote runs
instead require a provider termination receipt for every lease, with successful
cleanup and no active ownership. This applies to both per-turn and warm native
runners. A stop receipt retires only the settled cleanup owner for that exact company, run, provider, and sandbox resource, without changing its checkpoint or recorded action outcomes. Independent remote sandboxes have separate cleanup gates, including when one run owns multiple sandboxes. Successful pending-cleanup retries persist the same receipt and reconsider deferred user messages; a delivery failure never reverts successful provider cleanup. A failed checkpoint does not prevent destruction of a terminal run's isolated sandbox; busy ownership still prevents it.
Missing receipts and failed cleanup retain the hold. Older providers that return
no receipt remain supported but cannot authorize remote continuation. A terminal
database status or a PID check on the wrong host is insufficient.
No historical task is automatically awakened by this change.
### Explicit Recovery Action

4
packages/paperclip-runner/package.json
View File

@ -125,8 +125,8 @@
"report:runner-protocol-eval:catalog": "node scripts/runner-protocol-eval-campaign.mjs catalog",
"report:runner-protocol-eval:publish": "node scripts/publish-runner-protocol-eval-history.mjs",
"report:runner-chaos-evals": "pnpm run build:typescript && node scripts/run-runner-live-eval-schedule.mjs --mode chaos",
"check:conformance-parity": "cargo test --manifest-path runner/Cargo.toml --locked -p paperclip-runner-core runs_the_mock_core_path_with_stable_output",
"check:replay-parity": "cargo test --manifest-path runner/Cargo.toml --locked -p paperclip-runner-core replay_fixture_parity",
"check:conformance-parity": "cargo test --release --manifest-path runner/Cargo.toml --locked -p paperclip-runner-core runs_the_mock_core_path_with_stable_output",
"check:replay-parity": "cargo test --release --manifest-path runner/Cargo.toml --locked -p paperclip-runner-core replay_fixture_parity",
"docs:validate": "node scripts/validate-doc-links.mjs",
"trace:conformance": "pnpm run trace:conformance:rust",
"trace:conformance:rust": "cargo run --quiet --manifest-path runner/Cargo.toml --locked -p paperclip-runner-core --bin conformance-tracer",

54
packages/paperclip-runner/src/native-session-runtime.test.ts
View File

@ -27,6 +27,7 @@ import {
} from "./contracts/runtime-context.js";
import {
executeNativeSession,
completeTerminatedRemoteNativeSessionCleanup,
type ExecuteNativeSessionOptions,
} from "./native-session-runtime.js";
@ -3857,6 +3858,59 @@ describe("executeNativeSession recovery", () => {
},
);
it("retires only the terminated remote resource, including two sandboxes for one run", async () => {
const scopedIdentity = { ...identity, companyId: "remote-stop-company", runId: "remote-stop-run" };
const binding = { ...scopedIdentity, remoteCleanupScope: "first-sandbox" };
const scopedInput = { ...input, binding: { ...input.binding, companyId: scopedIdentity.companyId, runId: scopedIdentity.runId } };
const failure = new NativeSessionCloseUnrecoverableError();
const capabilities = { resume: true, typedEvents: true, steering: false, interruption: false, structuredResult: true };
const session: NativeSession = {
identity: () => scopedIdentity,
capabilities: async () => capabilities,
async *events() { throw new Error("cancelled remote transport"); },
startTurn: async () => ({ turnId: "remote-turn" }),
result: async () => null,
close: vi.fn(async () => { throw failure; }),
};
const backend: NativeSessionBackend = {
descriptor: async () => ({ kind: "remote", name: "remote-stop-test", version: "1", capabilities }),
openSession: vi.fn(async () => session),
};
const controlPlane: ControlPlanePort = {
openRun: async () => {}, checkpointSession: async () => {},
appendEvent: async () => ({ cursor: 0, highestContiguousSourceSeq: 0, disposition: "committed" }),
replayEvents: async () => ({ events: [], highestContiguousSourceSeq: 0 }),
completeRun: vi.fn(async () => {}),
};
const options = { input: scopedInput, backend, controlPlane, runnerInstanceId: "remote-runner",
controlPlaneInstanceId: "control", requireSessionCloseBeforeReturn: true,
remoteCleanupScope: binding.remoteCleanupScope };
await expect(executeNativeSession(options)).rejects.toBe(failure);
expect(completeTerminatedRemoteNativeSessionCleanup({ ...binding, runId: "other-run" })).toBe(true);
expect(completeTerminatedRemoteNativeSessionCleanup({ ...binding, companyId: "other-company" })).toBe(true);
await expect(executeNativeSession(options)).rejects.toBeInstanceOf(NativeSessionCleanupQuarantinedError);
expect(backend.openSession).toHaveBeenCalledOnce();
// A separate sandbox can start without inheriting this process quarantine.
const independent = { ...scopedIdentity, sessionId: "other-sandbox-session" };
const independentSession = { ...session, identity: () => independent };
const independentBackend = { ...backend, openSession: vi.fn(async () => independentSession) };
await expect(executeNativeSession({ ...options, remoteCleanupScope: "other-sandbox",
input: { ...scopedInput, binding: { ...scopedInput.binding, runId: independent.runId } },
backend: independentBackend })).rejects.toBe(failure);
expect(independentBackend.openSession).toHaveBeenCalledOnce();
expect(completeTerminatedRemoteNativeSessionCleanup(binding)).toBe(true);
// Same company/run, different sandbox: its quarantine must remain intact.
await expect(executeNativeSession({ ...options, remoteCleanupScope: "other-sandbox",
backend: independentBackend })).rejects.toBeInstanceOf(NativeSessionCleanupQuarantinedError);
expect(independentBackend.openSession).toHaveBeenCalledOnce();
// Reopening is now possible; the old failure/result was never rewritten.
await expect(executeNativeSession(options)).rejects.toBe(failure);
expect(backend.openSession).toHaveBeenCalledTimes(2);
expect(controlPlane.completeRun).not.toHaveBeenCalled();
completeTerminatedRemoteNativeSessionCleanup(binding);
completeTerminatedRemoteNativeSessionCleanup({ ...binding, remoteCleanupScope: "other-sandbox" });
});
it("propagates an exhausted required backend checkpoint close", async () => {
vi.useFakeTimers();
try {

32
packages/paperclip-runner/src/native-session-runtime.ts
View File

@ -125,6 +125,34 @@ export function completeRetainedNativeSessionCleanup(
return matches.length;
}
/** Control-plane-only cleanup boundary after the environment provider confirmed
* termination of the exact remote resource for this run. This retires process
* ownership, not checkpoints, action outcomes, or authorization to run again.
* An in-flight close must settle first: it must never reach a reused sandbox.
*/
export function completeTerminatedRemoteNativeSessionCleanup(binding: {
companyId: string;
runId: string;
remoteCleanupScope: string;
}): boolean {
if (!binding.remoteCleanupScope) return false;
const matches = [...quarantinedSessionCleanups].filter(({ session, domain }) => {
const identity = session.identity();
// Domains are created internally from company, backend kind/name, and the
// optional remote resource. Local domains have no fourth element.
const [, , , remoteCleanupScope] = JSON.parse(domain) as string[];
return identity.companyId === binding.companyId && identity.runId === binding.runId &&
remoteCleanupScope === binding.remoteCleanupScope;
});
if (matches.some(entry => entry.attempt || entry.recovery)) return false;
for (const entry of matches) {
if (entry.timer) clearTimeout(entry.timer);
entry.timer = null;
quarantinedSessionCleanups.delete(entry);
}
return true;
}
export interface NativeSessionGoalControl {
requestId: string;
action: "create" | "edit" | "replace" | "pause" | "resume" | "clear";
@ -138,6 +166,9 @@ export interface ExecuteNativeSessionOptions {
controlPlane: ControlPlanePort;
runnerInstanceId: string;
controlPlaneInstanceId: string;
/** Trusted provider resource identity: independent remote sandboxes must not
* inherit each other's process-cleanup gates. Omit for local backends. */
remoteCleanupScope?: string;
timeoutMs?: number;
/** Abort admission while waiting for prior cleanup in the same domain. */
signal?: AbortSignal;
@ -1739,6 +1770,7 @@ export async function executeNativeSession(
input.binding.companyId,
descriptor.kind,
descriptor.name,
...(options.remoteCleanupScope ? [options.remoteCleanupScope] : []),
]);
await retryQuarantinedSessionCleanups(cleanupDomain, options.signal);
if ("runtimeContext" in input) {

58
packages/plugins/sandbox-providers/daytona/src/plugin.test.ts
View File

@ -1291,13 +1291,38 @@ describe("Daytona sandbox provider plugin", () => {
expect(sandbox.process.executeCommand).toHaveBeenCalledTimes(1);
});
it("refreshes a cached stopped handle before granting a termination receipt", async () => {
process.env.DAYTONA_API_KEY = "host-key";
const sandbox = createMockSandbox({ id: "sandbox-resumed", state: "stopped" });
sandbox.refreshData.mockImplementation(async () => { sandbox.state = "started"; });
mockGet.mockResolvedValue(sandbox);
await expect(plugin.definition.onEnvironmentReleaseLease?.({
driverKey: "daytona", companyId: "company-1", environmentId: "env-1",
providerLeaseId: sandbox.id, config: { reuseLease: true },
})).resolves.toEqual({ providerLeaseId: sandbox.id, state: "stopped" });
expect(sandbox.refreshData).toHaveBeenCalled();
expect(sandbox.stop).toHaveBeenCalled();
});
it("does not acknowledge termination when both provider stop and delete fail", async () => {
process.env.DAYTONA_API_KEY = "host-key";
const sandbox = createMockSandbox({ id: "sandbox-failed-stop", state: "started" });
sandbox.stop.mockRejectedValueOnce(new Error("stop failed"));
sandbox.delete.mockRejectedValueOnce(new Error("delete failed"));
mockGet.mockResolvedValue(sandbox);
await expect(plugin.definition.onEnvironmentReleaseLease?.({
driverKey: "daytona", companyId: "company-1", environmentId: "env-1",
providerLeaseId: sandbox.id, config: { reuseLease: true },
})).rejects.toThrow("delete failed");
});
it("stops reusable leases and deletes ephemeral leases on release", async () => {
process.env.DAYTONA_API_KEY = "host-key";
const reusable = createMockSandbox({ id: "sandbox-reusable" });
const ephemeral = createMockSandbox({ id: "sandbox-ephemeral" });
mockGet.mockResolvedValueOnce(reusable).mockResolvedValueOnce(ephemeral);
await plugin.definition.onEnvironmentReleaseLease?.({
const reusableReceipt = await plugin.definition.onEnvironmentReleaseLease?.({
driverKey: "daytona",
companyId: "company-1",
environmentId: "env-1",
@ -1307,7 +1332,7 @@ describe("Daytona sandbox provider plugin", () => {
reuseLease: true,
},
});
await plugin.definition.onEnvironmentReleaseLease?.({
const ephemeralReceipt = await plugin.definition.onEnvironmentReleaseLease?.({
driverKey: "daytona",
companyId: "company-1",
environmentId: "env-1",
@ -1318,9 +1343,11 @@ describe("Daytona sandbox provider plugin", () => {
},
});
expect(reusableReceipt).toEqual({ providerLeaseId: "sandbox-reusable", state: "stopped" });
expect(ephemeralReceipt).toEqual({ providerLeaseId: "sandbox-ephemeral", state: "destroyed" });
expect(reusable.stop).toHaveBeenCalledWith(300);
expect(reusable.delete).not.toHaveBeenCalled();
expect(ephemeral.delete).toHaveBeenCalledWith(300);
expect(ephemeral.delete).toHaveBeenCalledWith(300, true);
});
it("archives instead of deleting when the lease was acquired with archiveOnRelease", async () => {
@ -1367,7 +1394,7 @@ describe("Daytona sandbox provider plugin", () => {
expect(sandbox.stop).not.toHaveBeenCalled();
expect(sandbox.archive).toHaveBeenCalled();
expect(sandbox.delete).toHaveBeenCalledWith(300);
expect(sandbox.delete).toHaveBeenCalledWith(300, true);
expect(warnSpy).toHaveBeenCalled();
});
@ -1389,7 +1416,7 @@ describe("Daytona sandbox provider plugin", () => {
});
expect(errored.stop).toHaveBeenCalledWith(300);
expect(errored.delete).toHaveBeenCalledWith(300);
expect(errored.delete).toHaveBeenCalledWith(300, true);
});
it("falls back to delete when stopping a healthy reusable lease fails mid-call", async () => {
@ -1411,7 +1438,7 @@ describe("Daytona sandbox provider plugin", () => {
});
expect(sandbox.stop).toHaveBeenCalledWith(300);
expect(sandbox.delete).toHaveBeenCalledWith(300);
expect(sandbox.delete).toHaveBeenCalledWith(300, true);
expect(warnSpy).toHaveBeenCalled();
});
@ -1628,7 +1655,24 @@ describe("Daytona sandbox provider plugin", () => {
expect(errorSpy).toHaveBeenCalledWith(expect.stringContaining(sessionId));
// The rest of teardown still ran: the ephemeral sandbox was deleted.
expect(sandbox.delete).toHaveBeenCalledWith(300);
expect(sandbox.delete).toHaveBeenCalledWith(300, true);
});
it("returns a destroy receipt only after the provider confirms deletion", async () => {
process.env.DAYTONA_API_KEY = "host-key";
const sandbox = createMockSandbox();
mockGet.mockResolvedValue(sandbox);
let complete!: () => void;
sandbox.delete.mockImplementationOnce(() => new Promise<void>(resolve => { complete = resolve; }));
const release = plugin.definition.onEnvironmentDestroyLease!({ driverKey: "daytona",
companyId: "company-1", environmentId: "env-1", providerLeaseId: "sandbox-123",
config: { timeoutMs: 300000, reuseLease: false } });
let settled = false;
void Promise.resolve(release).then(() => { settled = true; });
await vi.waitFor(() => expect(sandbox.delete).toHaveBeenCalledWith(300, true));
expect(settled).toBe(false);
complete();
await expect(release).resolves.toEqual({ providerLeaseId: "sandbox-123", state: "destroyed" });
});
it("clears the session store after delete so no orphan id survives a second teardown", async () => {

29
packages/plugins/sandbox-providers/daytona/src/plugin.ts
View File

@ -33,6 +33,7 @@ import type {
PluginEnvironmentRealizeWorkspaceParams,
PluginEnvironmentRealizeWorkspaceResult,
PluginEnvironmentReleaseLeaseParams,
PluginEnvironmentTerminationReceipt,
PluginEnvironmentResumeLeaseParams,
PluginEnvironmentStartInteractiveSetupParams,
PluginEnvironmentSyncInParams,
@ -2262,7 +2263,7 @@ const plugin = definePlugin({
async onEnvironmentReleaseLease(
params: PluginEnvironmentReleaseLeaseParams,
): Promise<void> {
): Promise<PluginEnvironmentTerminationReceipt | void> {
if (!params.providerLeaseId) return;
const config = parseDriverConfig(params.config);
const scope: SandboxScope = {
@ -2279,7 +2280,7 @@ const plugin = definePlugin({
sandboxHandleLeaseAdmissionStates.close(scope);
try {
const sandbox = await getSandboxOrNull(scope, { bypassTeardownGate: true });
if (!sandbox) return;
if (!sandbox) return { providerLeaseId: params.providerLeaseId, state: "destroyed" };
evictSandboxHandle(scope);
await sandboxHandleActivityGates.waitForIdle(scope);
@ -2288,6 +2289,9 @@ const plugin = definePlugin({
// so no channel outlives the sandbox and no stored channel id survives.
await closeDaytonaDuplexChannelsForLease(params.providerLeaseId);
// A cached stopped state is not a receipt: the resource could have been
// resumed since the handle was cached. Read provider state at this boundary.
await withLivenessTimeout("sandbox.refreshData", config.livenessTimeoutMs, () => sandbox.refreshData());
if (config.reuseLease) {
if (sandbox.state !== "stopped") {
try {
@ -2296,14 +2300,11 @@ const plugin = definePlugin({
console.warn(
`Failed to stop Daytona sandbox during lease release: ${formatErrorMessage(error)}. Attempting delete instead.`,
);
await sandbox.delete(toTimeoutSeconds(config.timeoutMs)).catch((deleteError) => {
console.warn(
`Failed to delete Daytona sandbox after stop failure: ${formatErrorMessage(deleteError)}`,
);
});
await sandbox.delete(toTimeoutSeconds(config.timeoutMs), true);
return { providerLeaseId: params.providerLeaseId, state: "destroyed" };
}
}
return;
return { providerLeaseId: params.providerLeaseId, state: "stopped" };
}
if (config.archiveOnRelease) {
@ -2313,7 +2314,7 @@ const plugin = definePlugin({
}
await sandbox.setAutoDeleteInterval(ARCHIVE_ON_RELEASE_AUTO_DELETE_MINUTES);
await sandbox.archive();
return;
return { providerLeaseId: params.providerLeaseId, state: "stopped" };
} catch (error) {
console.warn(
`Failed to archive Daytona sandbox during lease release: ${formatErrorMessage(error)}. Falling back to delete.`,
@ -2321,7 +2322,8 @@ const plugin = definePlugin({
}
}
await sandbox.delete(toTimeoutSeconds(config.timeoutMs));
await sandbox.delete(toTimeoutSeconds(config.timeoutMs), true);
return { providerLeaseId: params.providerLeaseId, state: "destroyed" };
} finally {
sandboxHandleTeardownGates.end(scope, teardownGate);
evictSandboxHandle(scope);
@ -2330,7 +2332,7 @@ const plugin = definePlugin({
async onEnvironmentDestroyLease(
params: PluginEnvironmentDestroyLeaseParams,
): Promise<void> {
): Promise<PluginEnvironmentTerminationReceipt | void> {
if (!params.providerLeaseId) return;
const config = parseDriverConfig(params.config);
const scope: SandboxScope = {
@ -2346,7 +2348,7 @@ const plugin = definePlugin({
sandboxHandleLeaseAdmissionStates.close(scope);
try {
const sandbox = await getSandboxOrNull(scope, { bypassTeardownGate: true });
if (!sandbox) return;
if (!sandbox) return { providerLeaseId: params.providerLeaseId, state: "destroyed" };
evictSandboxHandle(scope);
await sandboxHandleActivityGates.waitForIdle(scope);
@ -2354,7 +2356,8 @@ const plugin = definePlugin({
// Close every duplex channel on this lease before the delete, so no channel
// outlives the sandbox and no stored channel id survives.
await closeDaytonaDuplexChannelsForLease(params.providerLeaseId);
await sandbox.delete(toTimeoutSeconds(config.timeoutMs));
await sandbox.delete(toTimeoutSeconds(config.timeoutMs), true);
return { providerLeaseId: params.providerLeaseId, state: "destroyed" };
} finally {
sandboxHandleTeardownGates.end(scope, teardownGate);
evictSandboxHandle(scope);

5
packages/plugins/sdk/src/define-plugin.ts
View File

@ -75,6 +75,7 @@ import type {
PluginEnvironmentRealizeWorkspaceParams,
PluginEnvironmentRealizeWorkspaceResult,
PluginEnvironmentReleaseLeaseParams,
PluginEnvironmentTerminationReceipt,
PluginEnvironmentResumeLeaseParams,
PluginEnvironmentValidateConfigParams,
PluginEnvironmentValidationResult,
@ -384,12 +385,12 @@ export interface PluginDefinition {
/** Called when a run finishes and the provider lease can be released. */
onEnvironmentReleaseLease?(
params: PluginEnvironmentReleaseLeaseParams,
): Promise<void>;
): Promise<PluginEnvironmentTerminationReceipt | void>;
/** Called when the host needs to force-destroy provider state. */
onEnvironmentDestroyLease?(
params: PluginEnvironmentDestroyLeaseParams,
): Promise<void>;
): Promise<PluginEnvironmentTerminationReceipt | void>;
/** Called to materialize the run workspace inside the provider lease. */
onEnvironmentRealizeWorkspace?(

1
packages/plugins/sdk/src/index.ts
View File

@ -183,6 +183,7 @@ export type {
PluginEnvironmentAcquireLeaseParams,
PluginEnvironmentResumeLeaseParams,
PluginEnvironmentReleaseLeaseParams,
PluginEnvironmentTerminationReceipt,
PluginEnvironmentDestroyLeaseParams,
PluginEnvironmentRealizeWorkspaceParams,
PluginEnvironmentRealizeWorkspaceResult,

11
packages/plugins/sdk/src/protocol.ts
View File

@ -666,6 +666,13 @@ export interface PluginEnvironmentReleaseLeaseParams extends PluginEnvironmentDr
leaseMetadata?: Record<string, unknown>;
}
/** Returned only after the provider confirms that execution has ended. A queued
* stop request or successful local cleanup is not a termination receipt. */
export interface PluginEnvironmentTerminationReceipt {
providerLeaseId: string;
state: "stopped" | "destroyed";
}
export interface PluginEnvironmentDestroyLeaseParams extends PluginEnvironmentReleaseLeaseParams {}
export interface PluginEnvironmentRealizeWorkspaceParams extends PluginEnvironmentDriverBaseParams {
@ -1363,11 +1370,11 @@ export interface HostToWorkerMethods {
];
environmentReleaseLease: [
params: PluginEnvironmentReleaseLeaseParams,
result: void,
result: PluginEnvironmentTerminationReceipt | void,
];
environmentDestroyLease: [
params: PluginEnvironmentDestroyLeaseParams,
result: void,
result: PluginEnvironmentTerminationReceipt | void,
];
environmentRealizeWorkspace: [
params: PluginEnvironmentRealizeWorkspaceParams,

9
packages/plugins/sdk/src/testing.ts
View File

@ -55,6 +55,7 @@ import type {
PluginEnvironmentAcquireLeaseParams,
PluginEnvironmentResumeLeaseParams,
PluginEnvironmentReleaseLeaseParams,
PluginEnvironmentTerminationReceipt,
PluginEnvironmentDestroyLeaseParams,
PluginEnvironmentRealizeWorkspaceParams,
PluginEnvironmentRealizeWorkspaceResult,
@ -185,8 +186,8 @@ export interface EnvironmentTestHarnessOptions extends TestHarnessOptions {
onProbe?: (params: PluginEnvironmentProbeParams) => Promise<PluginEnvironmentProbeResult>;
onAcquireLease?: (params: PluginEnvironmentAcquireLeaseParams) => Promise<PluginEnvironmentLease>;
onResumeLease?: (params: PluginEnvironmentResumeLeaseParams) => Promise<PluginEnvironmentLease>;
onReleaseLease?: (params: PluginEnvironmentReleaseLeaseParams) => Promise<void>;
onDestroyLease?: (params: PluginEnvironmentDestroyLeaseParams) => Promise<void>;
onReleaseLease?: (params: PluginEnvironmentReleaseLeaseParams) => Promise<PluginEnvironmentTerminationReceipt | void>;
onDestroyLease?: (params: PluginEnvironmentDestroyLeaseParams) => Promise<PluginEnvironmentTerminationReceipt | void>;
onRealizeWorkspace?: (params: PluginEnvironmentRealizeWorkspaceParams) => Promise<PluginEnvironmentRealizeWorkspaceResult>;
onExecute?: (params: PluginEnvironmentExecuteParams) => Promise<PluginEnvironmentExecuteResult>;
onStartInteractiveSetup?: (params: PluginEnvironmentStartInteractiveSetupParams) => Promise<PluginEnvironmentInteractiveSetupSession>;
@ -210,9 +211,9 @@ export interface EnvironmentTestHarness extends TestHarness {
/** Invoke the environment driver's resumeLease hook. */
resumeLease(params: PluginEnvironmentResumeLeaseParams): Promise<PluginEnvironmentLease>;
/** Invoke the environment driver's releaseLease hook. */
releaseLease(params: PluginEnvironmentReleaseLeaseParams): Promise<void>;
releaseLease(params: PluginEnvironmentReleaseLeaseParams): Promise<PluginEnvironmentTerminationReceipt | void>;
/** Invoke the environment driver's destroyLease hook. */
destroyLease(params: PluginEnvironmentDestroyLeaseParams): Promise<void>;
destroyLease(params: PluginEnvironmentDestroyLeaseParams): Promise<PluginEnvironmentTerminationReceipt | void>;
/** Invoke the environment driver's realizeWorkspace hook. */
realizeWorkspace(params: PluginEnvironmentRealizeWorkspaceParams): Promise<PluginEnvironmentRealizeWorkspaceResult>;
/** Invoke the environment driver's execute hook. */

16
server/src/__tests__/environment-runtime.test.ts
View File

@ -1133,7 +1133,7 @@ describeEmbeddedPostgres("environmentRuntimeService", () => {
};
}
if (method === "environmentDestroyLease") {
return undefined;
return { providerLeaseId: "plugin-lease-1", state: "destroyed" };
}
throw new Error(`Unexpected plugin method: ${method}`);
}),
@ -1165,6 +1165,9 @@ describeEmbeddedPostgres("environmentRuntimeService", () => {
expect(leaseRows).toHaveLength(1);
expect(leaseRows[0]?.status).toBe("expired");
expect(leaseRows[0]?.cleanupStatus).toBe("success");
expect(leaseRows[0]?.metadata?.remoteExecutionTermination).toMatchObject({
companyId, runId, leaseId: leaseRows[0]!.id, providerLeaseId: "plugin-lease-1", state: "destroyed",
});
// The acquire provisioned the remote plugin sandbox, so it destroys the
// sandbox on the rejection. Without this teardown the rejected insert leaks a
@ -3014,7 +3017,12 @@ describeEmbeddedPostgres("environmentRuntimeService", () => {
const runtimeWithPlugin = environmentRuntimeService(db, { pluginWorkerManager: workerManager });
const lease = await environmentService(db).getLeaseById(orphan.id);
await runtimeWithPlugin.retryPendingSandboxTeardown({ environment: null, lease: lease! });
vi.mocked(workerManager.call).mockImplementationOnce(async (_pluginId, _method, args: any) => {
destroyConfigs.push(args.config);
return { providerLeaseId: args.providerLeaseId, state: "destroyed" };
});
await expect(runtimeWithPlugin.retryPendingSandboxTeardown({ environment: null, lease: lease! }))
.resolves.toEqual({ providerLeaseId: orphan.providerLeaseId, state: "destroyed" });
expect(destroyConfigs).toHaveLength(1);
// The recorded secret ref resolved to the old credential, and the resolved
// value reached the provider teardown instead of the secret ref.
@ -6218,7 +6226,7 @@ describeEmbeddedPostgres("environmentRuntimeService", () => {
isRunning: vi.fn((id: string) => id === pluginId),
call: vi.fn(async (_pluginId: string, method: string) => {
if (method === "environmentDestroyLease") {
return undefined;
return { providerLeaseId: reusableLease.providerLeaseId, state: "destroyed" };
}
throw new Error(`Unexpected plugin method: ${method}`);
}),
@ -6245,6 +6253,8 @@ describeEmbeddedPostgres("environmentRuntimeService", () => {
status: "expired",
failureReason: "environment_deleted",
cleanupStatus: "success",
metadata: { remoteExecutionTermination: { schema: "paperclip.remote-termination.v1",
leaseId: reusableLease.id, runId, providerLeaseId: reusableLease.providerLeaseId, state: "destroyed" } },
});
});

19
server/src/__tests__/heartbeat-run-terminalize-before-release.test.ts
View File

@ -7,6 +7,8 @@ import {
createDb,
heartbeatRunEvents,
heartbeatRuns,
environmentLeases,
environments,
issues,
} from "@paperclipai/db";
import {
@ -299,6 +301,23 @@ describeEmbeddedPostgres("heartbeat teardown terminalizes the run before releasi
expect(releaseRunLeases).not.toHaveBeenCalled();
});
it.each(["daytona", "local"])("destroy after failed checkpoint requires a terminal remote owner: %s", async provider => {
const { companyId, agentId, runId } = await seed({ issueStatus: "blocked", runStatus: "running" });
await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date() }).where(eq(heartbeatRuns.id, runId));
const [environment] = await db.insert(environments).values({ name: `cleanup-${runId}`, driver: "sandbox" }).returning();
await db.insert(environmentLeases).values({ companyId, environmentId: environment.id,
heartbeatRunId: runId, agentId, provider, providerLeaseId: "sandbox-owned",
status: "active", leasePolicy: "ephemeral" });
const releaseRunLeases = vi.fn(async () => []);
const heartbeat = heartbeatService(db, {
environmentRuntime: { releaseRunLeases } as unknown as HeartbeatEnvironmentRuntime,
closeWarmNativeSessionsForRun: async () => ({ closed: 0, busy: 0, failed: 1 }),
});
await heartbeat.releaseEnvironmentLeasesForRun({ runId, companyId, agentId,
status: "cancelled", providerResourceDisposition: "destroy" });
expect(releaseRunLeases).toHaveBeenCalledTimes(provider === "daytona" ? 1 : 0);
});
it("terminalizes a running run to succeeded before release when the issue reached done", async () => {
const { companyId, agentId, issueId, runId } = await seed({ issueStatus: "done", runStatus: "running" });

35
server/src/services/environment-runtime.ts
View File

@ -1,3 +1,4 @@
import { remoteTerminationReceipt } from "./remote-execution-termination.js";
import { createHash, randomUUID } from "node:crypto";
import { and, eq, inArray, sql } from "drizzle-orm";
import type { Db } from "@paperclipai/db";
@ -632,9 +633,10 @@ export interface EnvironmentRuntimeDriver {
* current environment provider, so a provider change or an environment delete
* cannot strand the teardown. `environment` is null when a delete already
* removed the environment row. The method throws when the teardown fails, so
* the cleanup sweep keeps the row for a later retry.
* the cleanup sweep keeps the row for a later retry. Any returned provider
* receipt must be validated and persisted by the caller at lease release.
*/
retryPendingSandboxTeardown?(input: { environment: Environment | null; lease: EnvironmentLease }): Promise<void>;
retryPendingSandboxTeardown?(input: { environment: Environment | null; lease: EnvironmentLease }): Promise<unknown>;
/**
* Report whether the provider worker can run an orphan teardown now. A plugin
* sandbox provider worker can be briefly down during its own restart window.
@ -1475,9 +1477,12 @@ function createSandboxEnvironmentDriver(
const releaseCleanedUpOrphanRow = async (
leaseId: string,
diagnosticFields: Record<string, unknown>,
receipt: unknown,
): Promise<void> => {
try {
const lease = await environmentsSvc.getLeaseById(leaseId);
await environmentsSvc.releaseLease(leaseId, "expired", {
...(lease ? { remoteExecutionTermination: remoteTerminationReceipt(lease, receipt) } : {}),
cleanupStatus: "success",
failureReason: "acquire_rejected_teardown_succeeded",
});
@ -1553,13 +1558,14 @@ function createSandboxEnvironmentDriver(
record: DeferredOrphanCleanupRecord;
cause: unknown;
canTeardown: boolean;
teardown: () => Promise<void>;
teardown: () => Promise<unknown>;
}): Promise<void> => {
const durable = await tryWriteDurablePendingCleanup(input.record);
let teardownFailed = !input.canTeardown;
let receipt: unknown;
if (!teardownFailed) {
try {
await input.teardown();
receipt = await input.teardown();
} catch {
teardownFailed = true;
}
@ -1567,7 +1573,7 @@ function createSandboxEnvironmentDriver(
if (!teardownFailed) {
// The teardown removed the orphan, so drop the durable row if we wrote one.
if (durable.leaseId !== null) {
await releaseCleanedUpOrphanRow(durable.leaseId, orphanDiagnosticFields(input.record));
await releaseCleanedUpOrphanRow(durable.leaseId, orphanDiagnosticFields(input.record), receipt);
}
return;
}
@ -2133,7 +2139,7 @@ function createSandboxEnvironmentDriver(
cause: error,
canTeardown: pluginWorkerManager.isRunning(pluginProvider.resolved.plugin.id),
teardown: async () => {
await pluginWorkerManager.call(
return await pluginWorkerManager.call(
pluginProvider.resolved.plugin.id,
"environmentDestroyLease",
{
@ -2488,7 +2494,7 @@ function createSandboxEnvironmentDriver(
{ issueId: input.lease.issueId, heartbeatRunId: input.lease.heartbeatRunId },
);
const workerConfig = stripSandboxProviderEnvelope(config as SandboxEnvironmentConfig);
await pluginWorkerManager.call(
return await pluginWorkerManager.call(
pluginProvider.resolved.plugin.id,
"environmentDestroyLease",
{
@ -2505,7 +2511,6 @@ function createSandboxEnvironmentDriver(
},
resolvePluginSandboxRpcTimeoutMs(workerConfig),
);
return;
}
// Built-in provider path. Resolve the recorded config secrets through the
@ -3005,6 +3010,7 @@ function createSandboxEnvironmentDriver(
const providerKey = readString(metadata.provider);
let cleanupStatus: "success" | "failed" = "success";
let termination: ReturnType<typeof remoteTerminationReceipt>;
if (
pluginId &&
providerKey &&
@ -3017,7 +3023,7 @@ function createSandboxEnvironmentDriver(
lease: input.lease,
provider: providerKey,
});
await runLeaseReleaseWithRunParent(input.lease.id, () =>
const receipt = await runLeaseReleaseWithRunParent(input.lease.id, () =>
pluginWorkerManager.call(pluginId, "environmentReleaseLease", {
driverKey: providerKey,
companyId: input.lease.companyId,
@ -3028,6 +3034,7 @@ function createSandboxEnvironmentDriver(
leaseMetadata: metadata,
}, resolvePluginSandboxRpcTimeoutMs(stripSandboxProviderEnvelope(config as SandboxEnvironmentConfig))),
);
termination = remoteTerminationReceipt(input.lease, receipt);
} catch {
cleanupStatus = "failed";
}
@ -3056,6 +3063,7 @@ function createSandboxEnvironmentDriver(
return await environmentsSvc.releaseLease(input.lease.id, releaseStatus, {
failureReason,
cleanupStatus,
...(cleanupStatus === "success" && termination ? { remoteExecutionTermination: termination } : {}),
});
}
@ -3065,6 +3073,7 @@ function createSandboxEnvironmentDriver(
failureReason: string;
}): Promise<EnvironmentLease | null> {
let cleanupStatus: "success" | "failed" = "success";
let termination: ReturnType<typeof remoteTerminationReceipt>;
const metadata = input.lease.metadata ?? {};
try {
@ -3084,7 +3093,7 @@ function createSandboxEnvironmentDriver(
lease: input.lease,
provider: providerKey,
});
await runLeaseReleaseWithRunParent(input.lease.id, () =>
const receipt = await runLeaseReleaseWithRunParent(input.lease.id, () =>
pluginWorkerManager.call(pluginId, "environmentDestroyLease", {
driverKey: providerKey,
companyId: input.lease.companyId,
@ -3095,6 +3104,7 @@ function createSandboxEnvironmentDriver(
leaseMetadata: metadata,
}, resolvePluginSandboxRpcTimeoutMs(stripSandboxProviderEnvelope(config as SandboxEnvironmentConfig))),
);
termination = remoteTerminationReceipt(input.lease, receipt);
}
} else {
const metadataConfig = sandboxConfigFromLeaseMetadata(input.lease);
@ -3124,6 +3134,7 @@ function createSandboxEnvironmentDriver(
{
failureReason: input.failureReason,
cleanupStatus,
...(cleanupStatus === "success" && termination ? { remoteExecutionTermination: termination } : {}),
},
);
}
@ -3814,14 +3825,14 @@ export function environmentRuntimeService(
async retryPendingSandboxTeardown(input: {
environment: Environment | null;
lease: EnvironmentLease;
}): Promise<void> {
}): Promise<unknown> {
const driver = requireDriverKey(getLeaseDriverKey(input.lease, input.environment));
if (!driver.retryPendingSandboxTeardown) {
throw new Error(
`Environment driver "${driver.driver}" does not support orphan sandbox teardown.`,
);
}
await driver.retryPendingSandboxTeardown(input);
return await driver.retryPendingSandboxTeardown(input);
},
// Report whether the provider worker can run an orphan teardown now. The

6
server/src/services/environments.ts
View File

@ -1615,6 +1615,7 @@ export function environmentService(db: Db) {
options?: {
failureReason?: string;
cleanupStatus?: EnvironmentLeaseCleanupStatus;
remoteExecutionTermination?: Record<string, unknown>;
},
) => {
const now = new Date();
@ -1627,6 +1628,11 @@ export function environmentService(db: Db) {
updatedAt: now,
...(options?.failureReason !== undefined ? { failureReason: options.failureReason } : {}),
...(options?.cleanupStatus !== undefined ? { cleanupStatus: options.cleanupStatus } : {}),
// A later release without a receipt cannot reuse an earlier stop's
// authority (for example after a same-run lease resume).
metadata: options?.remoteExecutionTermination
? sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || ${JSON.stringify({ remoteExecutionTermination: options.remoteExecutionTermination })}::jsonb`
: sql`${environmentLeases.metadata} - 'remoteExecutionTermination'`,
})
.where(eq(environmentLeases.id, id))
.returning()

106
server/src/services/explicit-native-continuation.test.ts
View File

@ -1,15 +1,16 @@
import { remoteTerminationReceipt } from "./remote-execution-termination.js";
import { randomUUID } from "node:crypto";
import { and, eq } from "drizzle-orm";
import { beforeAll, afterAll, describe, it, expect } from "vitest";
import {
approvals, issueApprovals, issueThreadInteractions,
agentWakeupRequests, agents, companies, createDb, heartbeatRuns, issueComments, issueRecoveryActions,
agentWakeupRequests, agents, companies, createDb, heartbeatRunEvents, heartbeatRuns, issueComments, issueRecoveryActions,
issues, nativeRunFinalizations, environmentLeases, environments, issueRelations, issueTreeHolds, issueTreeHoldMembers,
} from "@paperclipai/db";
import { startEmbeddedPostgresTestDatabase, getEmbeddedPostgresTestSupport } from "../__tests__/helpers/embedded-postgres.js";
import { admitExplicitNativeContinuation } from "./explicit-native-continuation.js";
import { buildExecutionContinuation } from "./execution-continuation.js";
import { heartbeatService } from "./heartbeat.js";
import { heartbeatService, type HeartbeatEnvironmentRuntime } from "./heartbeat.js";
import { getExecutionBlocker } from "./execution-blocker.js";
const support = await getEmbeddedPostgresTestSupport();
(support.supported ? describe : describe.skip)("explicit native conversation continuation", () => {
@ -46,6 +47,107 @@ const support = await getEmbeddedPostgresTestSupport();
agentId: f.agentId, status: "queued", contextSnapshot: { issueId: f.issueId, previousRunId: result.previousRunId, forceFreshSession: true } });
return result;
});
it.each([true, false])("acknowledges a legacy remote Stop only after confirmed lease cleanup: %s", async confirmed => {
const f = await seed();
await db.update(agents).set({ adapterType: "claude_local" }).where(eq(agents.id, f.agentId));
await db.update(heartbeatRuns).set({ runtimeMode: "legacy", status: "cancelled", processPid: null,
resultJson: { executionCancellation: { state: "requested" } },
}).where(eq(heartbeatRuns.id, f.sourceRunId));
await db.insert(heartbeatRunEvents).values({ companyId: f.companyId, runId: f.sourceRunId,
agentId: f.agentId, seq: 1, eventType: "adapter.invoke", payload: { adapterType: "claude_local" } });
await db.update(issueRecoveryActions).set({ cause: "legacy_execution_requires_reconciliation" })
.where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
const [environment] = await db.insert(environments).values({ name: `Remote ${f.sourceRunId}`, driver: "sandbox" }).returning();
const identity = { id: randomUUID(), companyId: f.companyId, heartbeatRunId: f.sourceRunId,
provider: "daytona", providerLeaseId: "sandbox-legacy" };
await db.insert(environmentLeases).values({ ...identity, environmentId: environment.id,
status: "expired", leasePolicy: "ephemeral", releasedAt: new Date(), cleanupStatus: "success",
metadata: confirmed ? { remoteExecutionTermination: remoteTerminationReceipt(identity,
{ providerLeaseId: identity.providerLeaseId, state: "destroyed" }) } : {},
});
await heartbeatService(db).releaseEnvironmentLeasesForRun({ runId: f.sourceRunId,
companyId: f.companyId, agentId: f.agentId, status: "cancelled" });
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId));
expect(run.resultJson?.executionCancellation).toMatchObject({ state: confirmed ? "acknowledged" : "requested" });
if (confirmed) expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
else expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull();
});
it.each(["stopped", "destroyed", "missing", "wrong_lease", "cleanup_failed", "active"])(
"admits a remote native predecessor only with confirmed termination: %s", async kind => {
const f = await seed();
// This PID exists on the control-plane host. It must never be used to
// infer liveness of the identically numbered remote process.
await db.update(heartbeatRuns).set({ processPid: process.pid }).where(eq(heartbeatRuns.id, f.sourceRunId));
const [environment] = await db.insert(environments).values({ name: `Remote ${f.sourceRunId}`, driver: "sandbox" }).returning();
const identity = { id: randomUUID(), companyId: f.companyId, heartbeatRunId: f.sourceRunId,
provider: "daytona", providerLeaseId: "sandbox-1" };
const proof = remoteTerminationReceipt(identity, { providerLeaseId: "sandbox-1",
state: kind === "destroyed" ? "destroyed" : "stopped" });
await db.insert(environmentLeases).values({ ...identity, environmentId: environment.id,
issueId: f.issueId, status: kind === "active" ? "active" : "released", leasePolicy: "ephemeral",
releasedAt: kind === "active" ? null : new Date(), cleanupStatus: kind === "cleanup_failed" ? "failed" : "success",
metadata: kind === "missing" ? {} : { remoteExecutionTermination:
kind === "wrong_lease" ? { ...proof, providerLeaseId: "other-sandbox" } : proof },
});
const result = await admit(f);
if (["stopped", "destroyed"].includes(kind)) expect(result).toMatchObject({ previousRunId: f.sourceRunId });
else expect(result).toBeNull();
},
);
it.each([
{ runtime: "native", retry: false }, { runtime: "native", retry: true },
{ runtime: "legacy", retry: false }, { runtime: "legacy", retry: true },
])("resumes a user message after confirmed cleanup: %j", async ({ runtime, retry }) => {
const f = await seed();
if (runtime === "legacy") {
await db.update(agents).set({ adapterType: "claude_local" }).where(eq(agents.id, f.agentId));
await db.update(heartbeatRuns).set({ runtimeMode: "legacy", status: "cancelled", processPid: null,
resultJson: { executionCancellation: { state: "requested" } } }).where(eq(heartbeatRuns.id, f.sourceRunId));
await db.insert(heartbeatRunEvents).values({ companyId: f.companyId, runId: f.sourceRunId,
agentId: f.agentId, seq: 1, eventType: "adapter.invoke", payload: { adapterType: "claude_local" } });
await db.update(issueRecoveryActions).set({ cause: "legacy_execution_requires_reconciliation" })
.where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
}
// Keep the successor queued so this test never starts an actual provider.
await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" });
const [environment] = await db.insert(environments).values({ name: `pending-${f.sourceRunId}`, driver: "sandbox" }).returning();
const identity = { id: randomUUID(), companyId: f.companyId, heartbeatRunId: f.sourceRunId,
provider: "daytona", providerLeaseId: "pending-sandbox" };
await db.insert(environmentLeases).values({ ...identity, environmentId: environment.id, status: "active", leasePolicy: "ephemeral" });
const heartbeat = heartbeatService(db, retry ? { environmentRuntime: {
retryPendingSandboxTeardown: async () => ({ providerLeaseId: identity.providerLeaseId, state: "destroyed" }),
} as unknown as HeartbeatEnvironmentRuntime } : {});
await heartbeat.wakeup(f.agentId, { source: "automation", triggerDetail: "system", reason: "issue_commented",
requestedByActorType: "user", requestedByActorId: "board", payload: { issueId: f.issueId, commentId: f.commentId },
contextSnapshot: { issueId: f.issueId, wakeCommentId: f.commentId } });
const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId));
await heartbeat.resumeRemoteStopComments(source);
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull();
if (retry) {
await db.update(environmentLeases).set({ status: "pending_cleanup", cleanupStatus: "failed" })
.where(eq(environmentLeases.id, identity.id));
expect(await heartbeat.sweepPendingCleanupLeases()).toMatchObject({ destroyed: 1 });
const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, identity.id));
expect(lease.metadata?.remoteExecutionTermination).toMatchObject({ runId: f.sourceRunId, state: "destroyed" });
} else {
await db.update(environmentLeases).set({ status: "released", releasedAt: new Date(), cleanupStatus: "success",
metadata: { remoteExecutionTermination: remoteTerminationReceipt(identity,
{ providerLeaseId: identity.providerLeaseId, state: "stopped" }) } }).where(eq(environmentLeases.id, identity.id));
await heartbeat.releaseEnvironmentLeasesForRun({ runId: source.id, companyId: source.companyId,
agentId: source.agentId, status: source.status });
}
await heartbeat.resumeRemoteStopComments(source);
await heartbeat.resumeRemoteStopComments(source);
const runs = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued")));
expect(runs).toHaveLength(1);
if (runtime === "native") expect(runs[0].contextSnapshot).toMatchObject({ forceFreshSession: true, previousRunId: f.sourceRunId,
explicitUserContinuation: { commentId: f.commentId } });
else expect(runs[0].contextSnapshot).toMatchObject({ wakeCommentId: f.commentId });
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
});
it("queues the actual user wake with a fresh session and retained source context", async () => {
const f = await seed();
// Occupy this agent's only slot so this admission test never starts a provider.

29
server/src/services/explicit-native-continuation.ts
View File

@ -1,3 +1,5 @@
import { completeTerminatedRemoteNativeSessionCleanup } from "../vendor/paperclip-runner/index.js";
import { hasRemoteTerminationReceipt, remoteLeaseCleanupScope } from "./remote-execution-termination.js";
import { z } from "zod";
import { and, eq, inArray, isNull, ne, or, sql } from "drizzle-orm";
import {
@ -76,24 +78,31 @@ export async function admitExplicitNativeContinuation(input: {
run.errorCode === "execution_reconciliation_required" &&
!run.processPid && !run.processGroupId && !run.nativeSessionId;
if (run.runtimeMode !== "native" && !unusedAdmission) return null;
if (!unusedAdmission) {
// A missing process identity is not evidence that a provider exited.
if (!run.processPid && !run.processGroupId) return null;
if (run.processPid && !processStopped(run.processPid)) return null;
if (run.processGroupId && !processStopped(-run.processGroupId)) return null;
}
const [coordinator] = await db.select().from(nativeRunFinalizations).where(and(
eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, run.id),
)).for("update");
if (coordinator && (coordinator.phase !== "terminal_failure" || coordinator.leaseOwner ||
coordinator.resultId || coordinator.failureDetail?.successorRunId)) return null;
const leases = await db.select({ provider: environmentLeases.provider, releasedAt: environmentLeases.releasedAt })
const leases = await db.select()
.from(environmentLeases).where(and(
eq(environmentLeases.companyId, companyId), eq(environmentLeases.heartbeatRunId, run.id),
));
// A PID on another host cannot be checked with this server's process table.
// Remote execution retains its hold until a target-aware stop proof exists.
if (leases.some(lease => !lease.releasedAt || lease.provider !== "local")) return null;
const remote = leases.some(lease => lease.provider !== "local");
if (remote) {
// Never interpret remote PIDs using the control-plane host's process table.
if (!leases.every(hasRemoteTerminationReceipt)) return null;
if (!input.dryRun && !leases.every(lease => completeTerminatedRemoteNativeSessionCleanup({
companyId, runId: run.id, remoteCleanupScope: remoteLeaseCleanupScope(lease)!,
}))) return null;
} else {
if (leases.some(lease => !lease.releasedAt || lease.cleanupStatus === "failed")) return null;
if (!unusedAdmission) {
// A missing process identity is not evidence that a provider exited.
if (!run.processPid && !run.processGroupId) return null;
if (run.processPid && !processStopped(run.processPid)) return null;
if (run.processGroupId && !processStopped(-run.processGroupId)) return null;
}
}
sources.push(run);
}
const nativeSources = sources.filter(run => run.runtimeMode === "native");

106
server/src/services/heartbeat.ts
View File

@ -1,3 +1,5 @@
import { completeTerminatedRemoteNativeSessionCleanup } from "../vendor/paperclip-runner/index.js";
import { remoteExecutionHasStopped, remoteTerminationReceipt, stoppedRemoteCleanupScopes } from "./remote-execution-termination.js";
import { admitExplicitNativeContinuation } from "./explicit-native-continuation.js";
import { getExecutionBlocker } from "./execution-blocker.js";
import { CONVERSATION_CONTINUATION_POLICY, runUsedConversationAdapter, hasConversationContinuationPolicy, isConversationAdapter } from "./conversation-continuation.js";
@ -9871,7 +9873,19 @@ export function heartbeatService(
);
return { closed: 0, busy: 0, failed: 1 };
});
if (closeResult.busy > 0 || closeResult.failed > 0) {
// A failed remote checkpoint cannot veto destruction of the isolated
// sandbox after the run stopped. Provider destruction supplies the exit
// proof; it does not turn the interrupted checkpoint into a success.
const remoteLeases = closeResult.failed > 0 && leaseOwnerRun &&
["cancelled", "failed", "timed_out", "interrupted"].includes(leaseOwnerRun.status)
? await db.select({ provider: environmentLeases.provider }).from(environmentLeases).where(and(
eq(environmentLeases.companyId, input.companyId),
eq(environmentLeases.heartbeatRunId, input.runId),
))
: [];
const canDestroyRemote = remoteLeases.length > 0 &&
remoteLeases.every(lease => lease.provider && lease.provider !== "local");
if (closeResult.busy > 0 || (closeResult.failed > 0 && !canDestroyRemote)) {
logger.warn(
{ runId: input.runId, warmNativeSessions: closeResult },
"deferred environment lease destruction until warm native sessions close",
@ -9906,6 +9920,76 @@ export function heartbeatService(
"failed to release environment lease for heartbeat run",
);
}
await acknowledgeRemoteStop(input.runId, input.companyId);
}
async function acknowledgeRemoteStop(runId: string, companyId: string) {
// The provider receipt arrives after adapter settlement. A remote ACP child
// has no host PID, so only this target-aware boundary can acknowledge Stop.
const stopped = await getRun(runId);
if (stopped?.runtimeMode === "native") {
const scopes = await stoppedRemoteCleanupScopes(db, companyId, runId);
for (const remoteCleanupScope of scopes ?? []) {
completeTerminatedRemoteNativeSessionCleanup({ companyId, runId, remoteCleanupScope });
}
}
if (stopped?.runtimeMode === "legacy" && stopped.status === "cancelled" &&
parseObject(stopped.resultJson?.executionCancellation).state === "requested" &&
await runUsedConversationAdapter(db, stopped) &&
await remoteExecutionHasStopped(db, companyId, runId)) {
await db.update(heartbeatRuns).set({
resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({
executionCancellation: { ...parseObject(stopped.resultJson?.executionCancellation),
state: "acknowledged", acknowledgedAt: new Date().toISOString(),
proof: "provider_termination_receipt" },
conversationContinuation: CONVERSATION_CONTINUATION_POLICY,
})}::jsonb`,
updatedAt: new Date(),
}).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, companyId),
eq(heartbeatRuns.status, "cancelled")));
}
}
async function resumeRemoteStopComments(run: typeof heartbeatRuns.$inferSelect) {
if (!isHeartbeatRunTerminalStatus(run.status) || adapterExecutionControls.has(run.id) ||
!(await remoteExecutionHasStopped(db, run.companyId, run.id))) return;
const issueId = run.nativeIssueId ?? (typeof run.contextSnapshot?.issueId === "string" ? run.contextSnapshot.issueId : null);
if (!issueId) return;
const legacyContinuation = run.runtimeMode === "legacy" && run.status === "cancelled" &&
hasConversationContinuationPolicy((await getRun(run.id))?.resultJson) &&
!(await getExecutionBlocker(db, run.companyId, issueId));
if (run.runtimeMode !== "native" && !legacyContinuation) return;
const pending = await db.select().from(agentWakeupRequests).where(and(
eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId),
eq(agentWakeupRequests.status, "deferred_issue_execution"),
eq(agentWakeupRequests.requestedByActorType, "user"),
sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`,
)).orderBy(asc(agentWakeupRequests.requestedAt));
for (const wake of pending) {
const payload = parseObject(wake.payload);
const context = parseObject(payload[DEFERRED_WAKE_CONTEXT_KEY]);
const commentId = deriveCommentId(context, payload);
if (legacyContinuation) {
if (!commentId || !run.finishedAt || !wake.requestedByActorId ||
!["issue_commented", "issue_reopened_via_comment"].includes(wake.reason ?? "")) continue;
const [comment] = await db.select().from(issueComments).where(and(
eq(issueComments.companyId, run.companyId), eq(issueComments.issueId, issueId),
sql`${issueComments.id}::text = ${commentId}`, eq(issueComments.authorType, "user"),
eq(issueComments.authorUserId, wake.requestedByActorId), isNull(issueComments.deletedAt),
isNull(issueComments.createdByRunId), gt(issueComments.createdAt, run.finishedAt),
));
if (!comment?.body.trim()) continue;
} else if (!await admitExplicitNativeContinuation({ db, companyId: run.companyId, issueId,
agentId: run.agentId, actorType: wake.requestedByActorType, actorId: wake.requestedByActorId,
reason: wake.reason, commentId, successorRunId: randomUUID(), dryRun: true })) continue;
// Re-enter ordinary admission with the original user's authority. It
// atomically adopts the deferred comments and still applies every gate.
await enqueueWakeup(run.agentId, { source: wake.source as WakeupOptions["source"], triggerDetail: (wake.triggerDetail ?? undefined) as WakeupOptions["triggerDetail"],
reason: wake.reason, payload, contextSnapshot: context,
requestedByActorType: "user", requestedByActorId: wake.requestedByActorId,
idempotencyKey: `remote-stop-comment:${run.id}:${wake.id}` });
break;
}
}
async function hasUnsafeTextProjectionDatabase() {
@ -17716,15 +17800,16 @@ export function heartbeatService(
try {
if (useRecordedTeardown) {
// Tear the sandbox down from the recorded provider config and the
// cleanup-authorized secret versions. The teardown returns no value
// and throws on failure, so the sweep releases the lease itself.
await environmentRuntime.retryPendingSandboxTeardown({
// cleanup-authorized secret versions. Preserve any provider receipt;
// a completed retry must grant the same evidence as initial cleanup.
const receipt = await environmentRuntime.retryPendingSandboxTeardown({
environment,
lease,
});
await environmentsSvc.releaseLease(lease.id, "expired", {
cleanupStatus: "success",
failureReason: "pending_cleanup_retry",
remoteExecutionTermination: remoteTerminationReceipt(lease, receipt),
});
destroyed += 1;
} else if (environment) {
@ -17761,6 +17846,15 @@ export function heartbeatService(
"pending_cleanup lease retry failed",
);
}
if (lease.heartbeatRunId) {
// Delivery failure must not revert successful provider cleanup. A new
// message can still use the persisted receipt on its next admission.
await (async () => {
await acknowledgeRemoteStop(lease.heartbeatRunId!, lease.companyId);
const run = await getRun(lease.heartbeatRunId!);
if (run) await resumeRemoteStopComments(run);
})().catch(() => logger.warn({ leaseId: lease.id }, "could not reconsider messages after cleanup retry"));
}
}
return { swept: rows.length, destroyed, capped };
@ -24720,6 +24814,9 @@ export function heartbeatService(
!nativeWorkspaceFinalizeScheduled &&
!shutdownInProgress
) {
if (latestRun) await resumeRemoteStopComments(latestRun).catch(err => {
logger.warn({ err, runId: run.id }, "failed to resume user messages after remote Stop");
});
await startNextQueuedRunForAgent(run.agentId);
}
}
@ -27924,6 +28021,7 @@ export function heartbeatService(
terminalizeRunOnLeaseRelease,
releaseEnvironmentLeasesForRun,
resumeRemoteStopComments,
sweepStaleIssueLocks,

10
server/src/services/native-runtime/native-session-executor.ts
View File

@ -1,3 +1,4 @@
import { remoteLeaseCleanupScope } from "../remote-execution-termination.js";
import {
boundedExecutionCleanup,
EXECUTION_CONTROL_DEADLINE_MS,
@ -7609,6 +7610,14 @@ async function executePaperclipNativeSessionWithinScope(
trace,
})
: null;
const remoteCleanupLease = input.runnerExecutionTarget?.kind === "remote" &&
input.runnerExecutionTarget.transport === "sandbox" && input.runnerExecutionTarget.leaseId
? await input.db.select({ provider: environmentLeases.provider, providerLeaseId: environmentLeases.providerLeaseId })
.from(environmentLeases).where(and(
eq(environmentLeases.companyId, input.execution.binding.companyId),
eq(environmentLeases.id, input.runnerExecutionTarget.leaseId),
)).then(rows => rows[0])
: null;
nativeSessionExecuteStartedAtMs = Date.now();
native = await trace.measure(
"native.session.execute",
@ -7621,6 +7630,7 @@ async function executePaperclipNativeSessionWithinScope(
const result = await trace.run(runnerSessionStartupScope, () =>
executeNativeSession({
input: runnerExecution,
remoteCleanupScope: remoteCleanupLease ? remoteLeaseCleanupScope(remoteCleanupLease) : undefined,
backend:
input.backend ??
runnerdBackend ??

52
server/src/services/remote-execution-termination.ts Normal file
View File

@ -0,0 +1,52 @@
import { and, eq } from "drizzle-orm";
import { environmentLeases, type Db } from "@paperclipai/db";
type LeaseIdentity = {
id: string; companyId: string; heartbeatRunId: string | null;
provider: string | null; providerLeaseId: string | null;
};
/** Bind a provider receipt to the exact host-owned lease and run. Old plugins
* return void; that remains supported but grants no continuation authority. */
export function remoteTerminationReceipt(lease: LeaseIdentity, value: unknown) {
const receipt = value as { providerLeaseId?: unknown; state?: unknown } | null;
if (!lease.heartbeatRunId || !lease.provider || lease.provider === "local" ||
!lease.providerLeaseId || receipt?.providerLeaseId !== lease.providerLeaseId ||
!["stopped", "destroyed"].includes(String(receipt?.state))) return undefined;
return {
schema: "paperclip.remote-termination.v1", companyId: lease.companyId,
runId: lease.heartbeatRunId, leaseId: lease.id, provider: lease.provider,
providerLeaseId: lease.providerLeaseId, state: receipt!.state,
confirmedAt: new Date().toISOString(),
};
}
export function hasRemoteTerminationReceipt(lease: LeaseIdentity & {
releasedAt: unknown; cleanupStatus: string | null; status: string;
metadata: Record<string, unknown> | null;
}): boolean {
const receipt = lease.metadata?.remoteExecutionTermination as Record<string, unknown> | undefined;
return Boolean(lease.releasedAt && lease.cleanupStatus === "success" &&
["released", "expired", "failed"].includes(lease.status) && receipt &&
receipt.schema === "paperclip.remote-termination.v1" &&
receipt.companyId === lease.companyId && receipt.runId === lease.heartbeatRunId &&
receipt.leaseId === lease.id && receipt.provider === lease.provider &&
remoteTerminationReceipt(lease, receipt));
}
export function remoteLeaseCleanupScope(lease: Pick<LeaseIdentity, "provider" | "providerLeaseId">) {
return lease.provider && lease.provider !== "local" && lease.providerLeaseId
? JSON.stringify([lease.provider, lease.providerLeaseId]) : undefined;
}
export async function stoppedRemoteCleanupScopes(db: Db, companyId: string, runId: string) {
const leases = await db.select().from(environmentLeases).where(and(
eq(environmentLeases.companyId, companyId), eq(environmentLeases.heartbeatRunId, runId),
));
if (leases.length === 0 || !leases.every(hasRemoteTerminationReceipt)) return null;
return [...new Set(leases.map(lease => remoteLeaseCleanupScope(lease)!))];
}
export async function remoteExecutionHasStopped(db: Db, companyId: string, runId: string) {
return await stoppedRemoteCleanupScopes(db, companyId, runId) !== null;
}

4
server/src/services/workspace-runtime-exposure.test.ts
View File

@ -356,7 +356,9 @@ function startInput(options?: {
command: options?.command ?? serviceCommand(),
env: { PAPERCLIP_PUBLIC_URL: "http://127.0.0.1:3100" },
port: options?.port ?? { type: "auto", envKey: "PORT" },
readiness: { type: "http", urlTemplate: "http://127.0.0.1:{{port}}", timeoutSec: 5 },
// These lifecycle tests spawn real servers; cold CI startup can take
// five seconds before the first listener is ready.
readiness: { type: "http", urlTemplate: "http://127.0.0.1:{{port}}", timeoutSec: 10 },
...(expose ? { expose } : {}),
}],
},

2
server/src/vendor/paperclip-runner/index.ts vendored
View File

@ -122,3 +122,5 @@ export const validatePrpStructuredRunResult =
runner.validatePrpStructuredRunResult;
export const NativeProviderTerminalFailure = runner.NativeProviderTerminalFailure;
export const completeTerminatedRemoteNativeSessionCleanup = runner.completeTerminatedRemoteNativeSessionCleanup;