fix: cancel stalled sandbox startup without waiting for setup (#13352)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Sandbox runs prepare credentials and files before an agent starts.
> - Stop must work during that preparation.
> - ACPX registered cancellation, but did not handle it while a setup
command was waiting.
> - Daytona cleanup waited for that same command before stopping the
sandbox.
> - This change stops the run's sandbox first and requires proof before
abandoning setup.

## Linked Issues or Issue Description

**What happened?**

Stop left a sandbox run active when remote credential setup stalled. The
run kept its connection lease until the sandbox was stopped separately.

**Expected behavior**

Stop terminates the selected run's sandbox, prevents later setup from
launching the agent, and lets run cleanup finish. It must not report
success without proof from the provider.

**Steps to reproduce**

1. Start an ACPX agent in Daytona.
2. Hold a command during remote credential or file setup.
3. Select Stop before the agent starts.
4. Before this fix, cleanup waits for the held command and never reaches
sandbox stop.

Related: #13351 exposed this during connection acceptance testing.
#12150 addresses scheduler load and session initialization limits, a
separate startup problem.

## What Changed

- Handle cancellation during ACPX sandbox preparation with a host-owned
stop callback.
- Pass an explicit active-work cancellation flag through environment
cleanup.
- Stop Daytona before draining setup commands. Keep normal graceful
cleanup.
- Require an exact run and lease termination receipt. Keep ownership of
outstanding requests when stop cannot be verified.
- Reject late setup work and defer sandbox resume until old requests
settle.
- Add regression tests and document the cancellation boundary.

## Verification

- Red: both the stalled ACPX setup test and the Daytona cancellation
test failed before the fix because Stop never reached the provider.
- Green: adapter and Daytona suites passed, along with cancellation
boundary and database-backed receipt/isolation tests.
- Two real Daytona probes ran a five-minute setup command. Cancellation
returned matching stopped receipts in 6.01 and 6.533 seconds. No later
setup command ran. Both test sandboxes were deleted.
- Repository typecheck and build passed locally. The full required CI
suite passed, including all server/workspace tests, browser shards,
Paperclip Runner verification, and canary packaging. The duplicate local
full-suite run was interrupted after CI passed; it is not claimed as a
completed local pass.
- No UI changes. The live probe uses the actual adapter cancellation
boundary and Daytona plugin; it is not a browser acceptance test.

## Risks

- Provider stop failures remain unacknowledged. The adapter keeps
ownership while its original requests remain active.
- A retry can receive a settling-work error until old provider requests
finish.
- Cancelled sandboxes are stopped and retained under their existing
provider expiry policy.
- Local execution and cancellation after the agent turn starts keep
their current behavior.
- Other providers must return an exact termination receipt to permit
early setup cancellation. No database migration.

## Model Used

OpenAI GPT-6 through Codex, with code execution and tool use. The exact
deployment identifier and context window 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 `#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 2026-09-13 08:35:43 -05:00
1 parent c9e3bb7ca4
commit 827ba8a434
14 files changed
+450 -18

No files matched your search

+14
View File
@@ -127,6 +127,20 @@ process may already hold credentials. The revoke confirmation lists attributed
active runs and exposes the existing Stop action; it does not promise immediate
provider-side revocation.
### Stop during sandbox preparation
ACPX startup registers cancellation while it materializes the remote auth home
and stages files. Stop requests termination of that run's sandbox. Daytona closes
admission and stops the sandbox before waiting for outstanding setup commands.
The host requires a receipt for the exact company, run, and provider lease before
abandoning the blocked setup RPC. Normal completion still drains work gracefully.
Late setup responses cannot launch the agent. A cancelled sandbox cannot resume
while its old provider requests are still settling; a retry receives an explicit
error instead. If termination cannot be verified, the adapter keeps ownership
until the outstanding operation settles, and Stop is not acknowledged as complete.
Local execution and cancellation of an already-running agent turn are unchanged.
## Legacy adoption
Migration `0273` indexes only explicitly owned personal secrets with a recognized
@@ -3793,6 +3793,63 @@ describe("ACPX engine remote session-lifecycle re-staging (PR 3: stage once / re
};
}
it("cancels a stalled remote startup without waiting for the command or launching the provider", async () => {
const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox();
const controller = new AbortController();
let entered!: () => void;
const started = new Promise<void>(resolve => { entered = resolve; });
let release!: () => void;
const blocked = new Promise<void>(resolve => { release = resolve; });
const stopRemoteStartup = vi.fn(async () => {});
const createRuntime = vi.fn(() => recordingRuntime({ ensureInputs: [] }) as never);
const originalExecute = executionTarget.runner.execute;
executionTarget.runner.execute = async input => {
if (input.command === "stalled-auth-setup") {
entered();
await blocked;
return { exitCode: 0, signal: null, timedOut: false, stdout: "", stderr: "", pid: null, startedAt: new Date().toISOString() };
}
return originalExecute(input);
};
const execute = createAcpxEngineExecutor({
createRuntime,
prepareRemoteManagedHome: async input => {
const target = input.executionTarget;
if (target?.kind !== "remote" || target.transport !== "sandbox" || !target.runner) {
throw new Error("Expected sandbox target");
}
await target.runner.execute({ command: "stalled-auth-setup" });
return { stagedRuntime: await input.stage([]) };
},
});
let settled = false;
const run = execute({
runId: "cancel-startup", runtime: {},
...baseExecuteArgs({ stateDir, localCwd, executionTarget }),
signal: controller.signal, stopRemoteStartup,
} as never).catch(error => error).finally(() => { settled = true; });
await started;
controller.abort(new Error("Stopped by user"));
try {
await vi.waitFor(() => expect(stopRemoteStartup).toHaveBeenCalledTimes(1), { timeout: 500 });
await vi.waitFor(() => expect(settled).toBe(true), { timeout: 500 });
expect(createRuntime).not.toHaveBeenCalled();
} finally {
release();
await run;
}
// A provider RPC completing late must not resume the cancelled startup.
expect(createRuntime).not.toHaveBeenCalled();
// A subsequent run can acquire the same local staging/auth preparation
// resources; cancellation must not leave the staging lease held.
const retry = await execute({
runId: "retry-after-cancel", runtime: {},
...baseExecuteArgs({ stateDir, localCwd, executionTarget }),
} as never);
expect(retry.exitCode).toBe(0);
expect(createRuntime).toHaveBeenCalledOnce();
});
it("test_acp_resume_compatible_session_does_not_restage", async () => {
const { stateDir, localCwd, remoteCwd, executionTarget } = await setupRemoteSandbox();
const ensureInputs: Array<Record<string, unknown>> = [];
@@ -1,3 +1,4 @@
import { cancellableSandboxStartup } from "./startup-cancellation.js";
import fs from "node:fs/promises";
import fsSync from "node:fs";
import os from "node:os";
@@ -4115,23 +4116,28 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
// run inside this wrap and overrides the store, so an in-step exec still
// parents to its step span. On a local or SSH target
// `spanParent.parentContext` is a no-op token, so the wrap is inert.
prepared = await runWithRuntimeParent(spanParent.parentContext, () =>
buildRuntime({
ctx,
engine,
deps,
ledger: runResourceLedger,
stagedIdleMs: warmIdleMs,
spanParent,
getRuntimeParentContext,
runtimeSpan: runRuntimeSpan,
stageRuntimeSpan: runStageSpan,
}),
);
buildRuntimeSettled = true;
// Capture the run's staging lease release now that the runtime built. The
// run root `finally` releases it as the final settlement act.
releaseStagingLease = prepared.sessionStagingLeaseRelease;
const startupCancellation = cancellableSandboxStartup(ctx);
try {
prepared = await runWithRuntimeParent(spanParent.parentContext, () =>
buildRuntime({
ctx: startupCancellation.context,
engine,
deps,
ledger: runResourceLedger,
stagedIdleMs: warmIdleMs,
spanParent,
getRuntimeParentContext,
runtimeSpan: runRuntimeSpan,
stageRuntimeSpan: runStageSpan,
}),
);
buildRuntimeSettled = true;
// Capture acquired resources before the cancellation boundary so the
// normal settlement path also releases a just-completed build.
releaseStagingLease = prepared.sessionStagingLeaseRelease;
} finally {
await startupCancellation.finish();
}
// Per-project staging outcomes for the referenced (mentioned) projects, surfaced back to the
// server on the run result. A referenced project that failed to stage into the sandbox is a
// first-class, counted failure in the requested-vs-synced observability, not only a warning. The
@@ -0,0 +1,124 @@
import { describe, expect, it, vi } from "vitest";
import type { AdapterExecutionContext } from "../types.js";
import type { CommandManagedDuplexChannel, CommandManagedRuntimeRunner } from "../command-managed-runtime.js";
import { cancellableSandboxStartup } from "./startup-cancellation.js";
function deferred<T>() {
let resolve!: (value: T) => void;
let reject!: (error: unknown) => void;
const promise = new Promise<T>((yes, no) => { resolve = yes; reject = no; });
return { promise, resolve, reject };
}
const result = { exitCode: 0, signal: null, timedOut: false, stdout: "", stderr: "" };
function fixture(stop = vi.fn(async () => {}), extra: Partial<CommandManagedRuntimeRunner> = {}) {
const controller = new AbortController();
const execute = vi.fn(async () => result);
const ctx = {
signal: controller.signal, stopRemoteStartup: stop,
executionTarget: { kind: "remote", transport: "sandbox", providerKey: "daytona", runner: { execute, ...extra } },
} as unknown as AdapterExecutionContext;
const startup = cancellableSandboxStartup(ctx);
const runner = (startup.context.executionTarget as { runner: CommandManagedRuntimeRunner }).runner;
return { controller, execute, stop, startup, runner };
}
describe("sandbox startup cancellation boundary", () => {
it("keeps ownership until termination is confirmed, including an early command error", async () => {
const receipt = deferred<void>();
const command = deferred<typeof result>();
const f = fixture(vi.fn(() => receipt.promise));
f.execute.mockReturnValue(command.promise);
let settled = false;
const outcome = f.runner.execute({ command: "setup" }).catch(error => error).finally(() => { settled = true; });
await vi.waitFor(() => expect(f.execute).toHaveBeenCalledOnce());
f.controller.abort(new Error("Stopped"));
command.reject(new Error("socket closed"));
await vi.waitFor(() => expect(f.stop).toHaveBeenCalledOnce());
expect(settled).toBe(false);
receipt.resolve();
expect(await outcome).toEqual(new Error("Stopped"));
await expect(f.startup.finish()).rejects.toThrow("Stopped");
await expect(f.runner.execute({ command: "late setup" })).rejects.toThrow("Stopped");
expect(f.execute).toHaveBeenCalledOnce();
});
it("does not abandon a hung command when the provider cannot confirm stop", async () => {
const command = deferred<typeof result>();
const f = fixture(vi.fn(async () => { throw new Error("stop unverified"); }));
f.execute.mockReturnValue(command.promise);
let settled = false;
const outcome = f.runner.execute({ command: "setup" }).catch(error => error).finally(() => { settled = true; });
await vi.waitFor(() => expect(f.execute).toHaveBeenCalledOnce());
f.controller.abort();
await vi.waitFor(() => expect(f.stop).toHaveBeenCalledOnce());
expect(settled).toBe(false);
command.resolve(result);
expect(await outcome).toEqual(new Error("stop unverified"));
await expect(f.startup.finish()).rejects.toThrow("stop unverified");
});
it("waits for all parallel setup requests after an unverified stop", async () => {
const first = deferred<typeof result>();
const second = deferred<typeof result>();
const f = fixture(vi.fn(async () => { throw new Error("stop unverified"); }));
f.execute.mockReturnValueOnce(first.promise).mockReturnValueOnce(second.promise);
let settled = false;
const a = f.runner.execute({ command: "first" }).catch(error => error).finally(() => { settled = true; });
const b = f.runner.execute({ command: "second" }).catch(error => error);
await vi.waitFor(() => expect(f.execute).toHaveBeenCalledTimes(2));
f.controller.abort();
first.reject(new Error("first failed"));
await vi.waitFor(() => expect(f.stop).toHaveBeenCalledOnce());
expect(settled).toBe(false);
second.resolve(result);
expect(await a).toEqual(new Error("stop unverified"));
expect(await b).toEqual(new Error("stop unverified"));
await expect(f.startup.finish()).rejects.toThrow("stop unverified");
});
it("disarms after successful startup and leaves ordinary turn cancellation alone", async () => {
const f = fixture();
await f.runner.execute({ command: "setup" });
await f.startup.finish();
f.controller.abort();
await expect(f.runner.execute({ command: "normal turn cleanup" })).resolves.toEqual(result);
expect(f.stop).not.toHaveBeenCalled();
expect(f.execute).toHaveBeenCalledTimes(2);
});
it("never starts a command when cancellation arrives before setup", async () => {
const f = fixture();
f.controller.abort(new Error("Stopped"));
await expect(f.runner.execute({ command: "setup" })).rejects.toThrow("Stopped");
await expect(f.startup.finish()).rejects.toThrow("Stopped");
expect(f.execute).not.toHaveBeenCalled();
expect(f.stop).toHaveBeenCalledOnce();
});
it("closes a duplex route returned after startup was cancelled", async () => {
const opening = deferred<CommandManagedDuplexChannel>();
const channel = {
write: vi.fn(), onData: vi.fn(), onExit: vi.fn(), stop: vi.fn(),
close: vi.fn(async () => {}),
};
const openDuplexChannel = vi.fn(() => opening.promise);
const f = fixture(vi.fn(async () => {}), { openDuplexChannel });
const outcome = f.runner.openDuplexChannel!({ command: ["agent"] }).catch(error => error);
await vi.waitFor(() => expect(openDuplexChannel).toHaveBeenCalledOnce());
f.controller.abort(new Error("Stopped"));
expect(await outcome).toEqual(new Error("Stopped"));
opening.resolve(channel);
await vi.waitFor(() => expect(channel.close).toHaveBeenCalledOnce());
await expect(f.startup.finish()).rejects.toThrow("Stopped");
});
it("does not stop another run's runner", async () => {
const a = fixture();
const b = fixture();
a.controller.abort(new Error("Stopped"));
await expect(a.startup.finish()).rejects.toThrow("Stopped");
await expect(b.runner.execute({ command: "setup" })).resolves.toEqual(result);
await b.startup.finish();
expect(b.stop).not.toHaveBeenCalled();
});
});
@@ -0,0 +1,88 @@
import type { AdapterExecutionContext } from "../types.js";
import type { CommandManagedRuntimeRunner } from "../command-managed-runtime.js";
/** Stop owns the sandbox before abandoning an outstanding startup RPC. The
* original RPC may complete late; its result cannot restart preparation. */
export function cancellableSandboxStartup(ctx: AdapterExecutionContext) {
const target = ctx.executionTarget;
const signal = ctx.signal;
const stop = ctx.stopRemoteStartup;
if (!signal || !stop || target?.kind !== "remote" || target.transport !== "sandbox" || !target.runner) {
return { context: ctx, finish: async () => {} };
}
let armed = true;
let stopping: Promise<void> | undefined;
const inFlight = new Set<Promise<unknown>>();
let rejectStopped!: (error: unknown) => void;
const stopped = new Promise<never>((_, reject) => { rejectStopped = reject; });
// Cancellation can arrive between RPCs, when nobody is awaiting this yet.
void stopped.catch(() => {});
const onAbort = () => {
if (stopping) return;
stopping = Promise.resolve().then(stop);
void stopping.then(
() => rejectStopped(signal.reason ?? new Error("Stopped during sandbox startup")),
// Without proof, the original operation still owns its resources. Do
// not abandon it or release credentials while it could be running.
() => {},
);
};
signal.addEventListener("abort", onAbort, { once: true });
if (signal.aborted) onAbort();
const joinStop = async () => {
try {
await stopping;
} catch (error) {
// Parallel setup may fail fast. Without a receipt, retain ownership of
// every outstanding RPC, not only the first one that returns an error.
await Promise.allSettled([...inFlight]);
throw error;
}
};
const guard = async <T>(call: () => Promise<T>): Promise<T> => {
if (!armed && !stopping) return call();
if (stopping) {
await joinStop();
return stopped;
}
const operation = Promise.resolve().then(() => {
if (stopping) signal.throwIfAborted();
return call();
});
inFlight.add(operation);
void operation.then(() => inFlight.delete(operation), () => inFlight.delete(operation));
try {
return await Promise.race([operation, stopped]);
} finally {
if (stopping) {
await joinStop();
signal.throwIfAborted();
}
}
};
const original = target.runner;
const runner: CommandManagedRuntimeRunner = {
...original,
execute: input => guard(() => original.execute(input)),
...(original.syncIn ? { syncIn: (input: Parameters<NonNullable<typeof original.syncIn>>[0]) => guard(() => original.syncIn!(input)) } : {}),
...(original.syncOut ? { syncOut: (input: Parameters<NonNullable<typeof original.syncOut>>[0]) => guard(() => original.syncOut!(input)) } : {}),
...(original.openDuplexChannel ? {
openDuplexChannel: (input: Parameters<NonNullable<typeof original.openDuplexChannel>>[0]) => guard(async () => {
const channel = await original.openDuplexChannel!(input);
// An open RPC can return after confirmed termination. Its host route
// must not survive just because the caller already abandoned the RPC.
if (stopping) await channel.close();
return channel;
}),
} : {}),
};
return {
context: { ...ctx, executionTarget: { ...target, runner } },
async finish() {
signal.removeEventListener("abort", onAbort);
armed = false;
if (stopping) await joinStop();
signal.throwIfAborted();
},
};
}
+3
View File
@@ -199,6 +199,9 @@ export interface AdapterExecutionContext {
signal?: AbortSignal;
/** Opt in to signal-based cancellation before starting provider work. */
onCancellationReady?: () => Promise<void>;
/** Host-owned stop of this run's sandbox during setup. Resolves only after
* provider termination is verified; never accepts an agent-selected lease. */
stopRemoteStartup?: () => Promise<void>;
/** Server-owned, actor-attributed snapshot also rendered by legacy wake prompts. */
executionContinuation?: ExecutionContinuationEnvelope | null;
runId: string;
@@ -2660,6 +2660,52 @@ describe("Daytona sandbox provider plugin", () => {
expect(sandbox.process.executeCommand).toHaveBeenCalledTimes(1);
});
it("cancels active work before waiting for a stalled execute to drain", async () => {
process.env.DAYTONA_API_KEY = "host-key";
const sandbox = createMockSandbox({ id: "lease-a" });
let release!: () => void;
sandbox.process.executeCommand.mockImplementation(async () => {
await new Promise<void>(resolve => { release = resolve; });
return { exitCode: 0, result: "", artifacts: { stdout: "" } };
});
mockGet.mockResolvedValue(sandbox);
const execute = plugin.definition.onEnvironmentExecute!(execParams("lease-a"));
await vi.waitFor(() => expect(release).toBeTypeOf("function"));
const cancellation = plugin.definition.onEnvironmentReleaseLease!({
driverKey: "daytona", companyId: "company-1", environmentId: "env-1",
providerLeaseId: "lease-a", config: { timeoutMs: 300000, reuseLease: true },
cancelActiveWork: true,
});
try {
await vi.waitFor(() => expect(sandbox.stop).toHaveBeenCalledTimes(1), { timeout: 500 });
await expect(cancellation).resolves.toEqual({ providerLeaseId: "lease-a", state: "stopped" });
await expect(plugin.definition.onEnvironmentExecute!(execParams("lease-a"))).rejects.toThrow(/no longer active/);
await expect(plugin.definition.onEnvironmentResumeLease!({
driverKey: "daytona", companyId: "company-1", environmentId: "env-1",
providerLeaseId: "lease-a", config: { timeoutMs: 300000, reuseLease: true },
})).rejects.toThrow(/still settling cancelled work/);
expect(sandbox.start).not.toHaveBeenCalled();
} finally {
release();
await execute;
await cancellation;
}
});
it("does not report termination when the provider rejects Stop", async () => {
process.env.DAYTONA_API_KEY = "host-key";
const sandbox = createMockSandbox({ id: "lease-a" });
sandbox.stop.mockRejectedValue(new Error("provider unavailable"));
mockGet.mockResolvedValue(sandbox);
await expect(plugin.definition.onEnvironmentReleaseLease!({
driverKey: "daytona", companyId: "company-1", environmentId: "env-1",
providerLeaseId: "lease-a", config: { timeoutMs: 300000, reuseLease: true },
cancelActiveWork: true,
})).rejects.toThrow("provider unavailable");
expect(sandbox.delete).not.toHaveBeenCalled();
await expect(plugin.definition.onEnvironmentExecute!(execParams("lease-a"))).rejects.toThrow(/no longer active/);
});
it("waits for an in-flight execute before teardown cleanup starts", async () => {
process.env.DAYTONA_API_KEY = "host-key";
const sandbox = createMockSandbox({ id: "lease-a" });
@@ -1128,7 +1128,11 @@ const sandboxHandleActivityGates = (() => {
gates.clear();
}
return { begin, waitForIdle, end, reset };
function isActive(scope: SandboxScope): boolean {
return gates.has(sandboxHandleCacheKey(scope));
}
return { begin, waitForIdle, isActive, end, reset };
})();
type SandboxLeaseAdmissionOptions = {
@@ -2189,6 +2193,11 @@ const plugin = definePlugin({
providerLeaseId: params.providerLeaseId,
config,
};
// A confirmed stop may precede completion of an old SDK request. Do not
// restart its sandbox under that request; it could still write or execute.
if (sandboxHandleLeaseAdmissionStates.isClosed(scope) && sandboxHandleActivityGates.isActive(scope)) {
throw new Error("The stopped Daytona sandbox is still settling cancelled work. Retry shortly.");
}
return await withSandboxActivityGate(scope, async () => {
const sandbox = await getSandboxOrNull(scope, { bypassTeardownGate: true });
if (!sandbox) {
@@ -2285,6 +2294,16 @@ const plugin = definePlugin({
if (!sandbox) return { providerLeaseId: params.providerLeaseId, state: "destroyed" };
evictSandboxHandle(scope);
if (params.cancelActiveWork) {
// Graceful release waits for activity below. Stop must not wait on the
// command it is cancelling. Admission is already closed for this lease.
const timeoutSeconds = Math.min(toTimeoutSeconds(config.timeoutMs), 30);
await withLivenessTimeout("sandbox.refreshData", Math.min(config.livenessTimeoutMs, 30_000), () => sandbox.refreshData());
if (sandbox.state !== "stopped") await sandbox.stop(timeoutSeconds);
sandboxHandleSessionStore.clear(scope);
await closeDaytonaDuplexChannelsForLease(params.providerLeaseId);
return { providerLeaseId: params.providerLeaseId, state: "stopped" };
}
await sandboxHandleActivityGates.waitForIdle(scope);
await teardownSession(sandbox, scope);
// Close every duplex channel on this lease before the stop or the delete,
+3
View File
@@ -662,6 +662,9 @@ export interface PluginEnvironmentResumeLeaseParams extends PluginEnvironmentDri
}
export interface PluginEnvironmentReleaseLeaseParams extends PluginEnvironmentDriverBaseParams {
/** Explicit operator cancellation: terminate active work instead of waiting
* for command/sync activity to drain. Still requires a provider receipt. */
cancelActiveWork?: boolean;
providerLeaseId: string | null;
leaseMetadata?: Record<string, unknown>;
}
@@ -255,6 +255,20 @@ describe("environmentRunOrchestrator — realizeForRun", () => {
mockLogActivity.mockResolvedValue(undefined);
});
it.each([false, true])("only requests active-work cancellation for explicit Stop: %s", async (cancelActiveWork) => {
const releaseRunLeases = vi.fn().mockResolvedValue([]);
const runtime = makeMockRuntime({ releaseRunLeases });
const orchestrator = environmentRunOrchestrator(mockDb, { environmentRuntime: runtime });
await orchestrator.releaseForRun({
heartbeatRunId: "run-1", companyId: "company-1", agentId: "agent-1",
providerResourceDisposition: "stop_and_retain", cancelActiveWork,
});
expect(releaseRunLeases).toHaveBeenCalledWith(
"run-1", "released", expect.any(Function), "stop_and_retain",
...(cancelActiveWork ? [true] : []),
);
});
it("happy path: returns lease, executionTarget, and remoteExecution on successful realization", async () => {
const executionTarget = { kind: "local", environmentId: "env-1", leaseId: "lease-1" };
const remoteExecution = { kind: "local", environmentId: "env-1", leaseId: "lease-1" };
@@ -39,6 +39,7 @@ import * as sandboxProviderRuntime from "../services/sandbox-provider-runtime.ts
import * as environmentsModule from "../services/environments.ts";
import { logger } from "../middleware/logger.ts";
import { environmentService } from "../services/environments.ts";
import { remoteExecutionHasStopped } from "../services/remote-execution-termination.ts";
import { heartbeatService } from "../services/heartbeat.ts";
import { secretService } from "../services/secrets.ts";
import type { PluginWorkerManager } from "../services/plugin-worker-manager.ts";
@@ -463,6 +464,34 @@ describeEmbeddedPostgres("environmentRuntimeService", () => {
return { pluginId, companyId, agentId, environment, runId, executionWorkspaceId, reusableLease };
}
it.each(["stopped", "missing", "wrong-lease", "error"])(
"startup cancellation is run-scoped and needs a matching receipt: %s",
async (outcome) => {
const { pluginId, companyId, runId, reusableLease, environment } = await seedReusablePluginSandboxLease();
const other = await environmentService(db).acquireLease({
companyId, environmentId: environment.id, heartbeatRunId: null,
leasePolicy: "ephemeral", provider: "fake-plugin", providerLeaseId: "other-sandbox",
});
const call = vi.fn(async () => {
if (outcome === "error") throw new Error("provider unavailable");
if (outcome === "missing") return undefined;
return { providerLeaseId: outcome === "wrong-lease" ? "other-sandbox" : reusableLease.providerLeaseId, state: "stopped" };
});
const workerManager = {
isRunning: () => true, call,
getWorker: () => ({ supportedMethods: ["environmentReleaseLease"] }),
} as unknown as PluginWorkerManager;
const runtimeWithPlugin = environmentRuntimeService(db, { pluginWorkerManager: workerManager });
await runtimeWithPlugin.releaseRunLeases(runId, "released", undefined, "stop_and_retain", true);
expect(call).toHaveBeenCalledWith(pluginId, "environmentReleaseLease", expect.objectContaining({
companyId, providerLeaseId: reusableLease.providerLeaseId, cancelActiveWork: true,
}), expect.any(Number));
expect(call).toHaveBeenCalledOnce();
expect(await remoteExecutionHasStopped(db, companyId, runId)).toBe(outcome === "stopped");
await expect(environmentService(db).getLeaseById(other.id)).resolves.toMatchObject({ status: "active" });
},
);
it("retains a successful reusable sandbox lease without stopping the provider resource", async () => {
const { pluginId, runId, reusableLease } = await seedReusablePluginSandboxLease();
const workerManager = {
@@ -584,6 +584,8 @@ export function environmentRunOrchestrator(
agentId: string;
status?: Extract<EnvironmentLeaseStatus, "released" | "expired" | "failed">;
failureReason?: string;
/** Explicit Stop during adapter startup; never used for ordinary cleanup. */
cancelActiveWork?: boolean;
/** Explicit paperclip_runner resource lifecycle. Omitted for legacy adapters. */
providerResourceDisposition?: ProviderResourceDisposition;
nativeLifecycleTelemetry?: {
@@ -606,6 +608,7 @@ export function environmentRunOrchestrator(
status,
(leaseId, error) => result.errors.push({ leaseId, error }),
input.providerResourceDisposition,
...(input.cancelActiveWork ? [true] as const : []),
);
} catch (err) {
result.errors.push({ leaseId: "*", error: err });
@@ -477,6 +477,8 @@ export interface EnvironmentDriverAcquireInput {
}
export interface EnvironmentDriverReleaseInput {
/** Explicit Stop may terminate in-flight setup rather than drain it. */
cancelActiveWork?: boolean;
environment: Environment;
lease: EnvironmentLease;
status: Extract<EnvironmentLeaseStatus, "released" | "expired" | "failed">;
@@ -3032,9 +3034,11 @@ function createSandboxEnvironmentDriver(
config: stripSandboxProviderEnvelope(config as SandboxEnvironmentConfig),
providerLeaseId: input.lease.providerLeaseId,
leaseMetadata: metadata,
...(input.cancelActiveWork ? { cancelActiveWork: true } : {}),
}, resolvePluginSandboxRpcTimeoutMs(stripSandboxProviderEnvelope(config as SandboxEnvironmentConfig))),
);
termination = remoteTerminationReceipt(input.lease, receipt);
if (input.cancelActiveWork && !termination) cleanupStatus = "failed";
} catch {
cleanupStatus = "failed";
}
@@ -3699,6 +3703,7 @@ export function environmentRuntimeService(
status: Extract<EnvironmentLeaseStatus, "released" | "expired" | "failed"> = "released",
onLeaseReleaseError?: (leaseId: string, error: unknown) => void,
providerResourceDisposition?: ProviderResourceDisposition,
cancelActiveWork?: boolean,
): Promise<EnvironmentRuntimeLeaseRecord[]> {
const leaseRows = await db
.select()
@@ -3781,6 +3786,7 @@ export function environmentRuntimeService(
})
: driver
? await driver.releaseRunLease({
...(cancelActiveWork ? { cancelActiveWork: true } : {}),
environment,
lease: leaseSnapshot,
// A stopped reusable provider resource must remain eligible
+20
View File
@@ -23724,6 +23724,26 @@ export function heartbeatService(
},
onDispatch: markDispatchStarted,
signal: executionControl.controller.signal,
...(executionTarget?.kind === "remote" && executionTarget.transport === "sandbox" ? {
stopRemoteStartup: async () => {
// Scope comes from the running host invocation, never agent
// config. Keep adapter ownership until setup has unwound.
if (!executionControl.controller.signal.aborted) {
throw new Error("Remote startup stop requires a cancelled run");
}
const release = await envOrchestrator.releaseForRun({
heartbeatRunId: run.id,
companyId: agent.companyId,
agentId: agent.id,
status: "released",
providerResourceDisposition: "stop_and_retain",
cancelActiveWork: true,
});
if (release.errors.length || !await remoteExecutionHasStopped(db, agent.companyId, run.id)) {
throw new Error("Could not verify remote startup stopped");
}
},
} : {}),
onCancellationReady: async () => {
await registerAdapterExecutionControl(run.id, executionControl);
const current = await getRun(run.id);