mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-26 02:00:52 +01:00
Fence legacy sandbox dispatch against startup cancellation
Reconcile persisted cancellation with newly registered scopes, check cancellation before dispatch, and preserve external child termination signals. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
066b4f0160
commit
1caad1bf9e
5 files changed
+84
-8
No files matched your search
@@ -6,6 +6,7 @@ import * as processes from "../services/hot-restart.js";
|
||||
import * as adapters from "../adapters/index.js";
|
||||
import * as orchestration from "../services/environment-run-orchestrator.js";
|
||||
import * as compatibility from "../services/legacy-sandbox-workspace.js";
|
||||
import * as cancellation from "@paperclipai/adapter-utils/adapter-run-cancellation";
|
||||
import { bindAdapterRunStop, hasAdapterRunCancellation } from "@paperclipai/adapter-utils/adapter-run-cancellation";
|
||||
import * as executionTargets from "@paperclipai/adapter-utils/execution-target";
|
||||
import { heartbeatService, persistHeartbeatRunProcessMetadata } from "../services/heartbeat.js";
|
||||
@@ -68,7 +69,41 @@ describe("heartbeat process identity persistence", () => {
|
||||
}
|
||||
}, 30_000);
|
||||
|
||||
it("cancels the remote adapter and waits for teardown before acknowledging", async () => {
|
||||
it("does not dispatch when cancellation precedes the remote cancellation scope", async () => {
|
||||
const originalOrchestrator = orchestration.environmentRunOrchestrator;
|
||||
vi.spyOn(orchestration, "environmentRunOrchestrator").mockImplementation((...args) => {
|
||||
const actual = originalOrchestrator(...args);
|
||||
return { ...actual, realizeForRun: async (input) => ({
|
||||
...await actual.realizeForRun(input),
|
||||
executionTarget: { kind: "remote", transport: "sandbox", remoteCwd: "/remote/task", shellCommand: "sh" } as never,
|
||||
}) };
|
||||
});
|
||||
vi.spyOn(compatibility, "hasLegacySandboxWorkspace").mockReturnValue(true);
|
||||
let ready!: () => void, release!: () => void;
|
||||
const preparing = new Promise<void>((resolve) => { ready = resolve; });
|
||||
const finishPreparation = new Promise<void>((resolve) => { release = resolve; });
|
||||
vi.spyOn(executionTargets, "prepareGitHubOperationLaunchers").mockImplementation(async (input) => {
|
||||
ready(); await finishPreparation; return input.env;
|
||||
});
|
||||
vi.spyOn(executionTargets, "cleanupGitHubOperationLaunchers").mockResolvedValue(undefined);
|
||||
const execute = vi.fn(async () => ({ exitCode: 0, signal: null, timedOut: false }));
|
||||
vi.spyOn(adapters, "getServerAdapter").mockReturnValue({ supportsLocalAgentJwt: false, execute } as never);
|
||||
const heartbeat = heartbeatService(db);
|
||||
try {
|
||||
const queued = await heartbeat.invoke(agentId, "on_demand", {}, "manual");
|
||||
expect(queued).not.toBeNull();
|
||||
await preparing;
|
||||
expect(hasAdapterRunCancellation(queued!.id)).toBe(false);
|
||||
await heartbeat.cancelRun(queued!.id);
|
||||
expect((await heartbeat.getRun(queued!.id))?.status).toBe("cancelled");
|
||||
release();
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
expect(execute).not.toHaveBeenCalled();
|
||||
expect(hasAdapterRunCancellation(queued!.id)).toBe(false);
|
||||
} finally { release(); await heartbeat.drainActiveRunExecutions(); }
|
||||
}, 30_000);
|
||||
|
||||
it.each([false, true])("cancels the remote adapter and waits for teardown (scope lookup raced: %s)", async (scopeLookupRaced) => {
|
||||
const originalOrchestrator = orchestration.environmentRunOrchestrator;
|
||||
vi.spyOn(orchestration, "environmentRunOrchestrator").mockImplementation((...args) => {
|
||||
const actual = originalOrchestrator(...args);
|
||||
@@ -105,6 +140,9 @@ describe("heartbeat process identity persistence", () => {
|
||||
const queued = await heartbeat.invoke(agentId, "on_demand", {}, "manual");
|
||||
expect(queued).not.toBeNull();
|
||||
await started;
|
||||
// Model the initial lookup occurring before registration. The final
|
||||
// lookup after persisting cancellation must still notify the scope.
|
||||
if (scopeLookupRaced) vi.spyOn(cancellation, "hasAdapterRunCancellation").mockReturnValueOnce(false);
|
||||
let acknowledged = false;
|
||||
pending = heartbeat.cancelRun(queued!.id).then((result) => { acknowledged = true; return result; });
|
||||
await interrupted;
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { beginAdapterRunCancellation, cancelAdapterRunExecution, finishAdapterRunCancellation, hasAdapterRunCancellation } from "@paperclipai/adapter-utils/adapter-run-cancellation";
|
||||
import { beginAdapterRunCancellation, cancelAdapterRunExecution, finishAdapterRunCancellation, hasAdapterRunCancellation, throwIfAdapterRunCancelled } from "@paperclipai/adapter-utils/adapter-run-cancellation";
|
||||
import { measureSandboxOperation, runWithSandboxPerformanceTrace, setSandboxPerformanceRunAttributes } from "./sandbox-performance.js";
|
||||
import { initializeRunIdentity } from "./run-identity.js";
|
||||
import { startNativeGitHubCallbackBridge } from "./native-github-bridge.js";
|
||||
@@ -21151,6 +21151,11 @@ export function heartbeatService(
|
||||
if (nativeRuntimeResolution.kind === "legacy"
|
||||
&& executionTarget?.kind === "remote" && executionTarget.transport === "sandbox") {
|
||||
beginAdapterRunCancellation(run.id);
|
||||
// Register first, then read the durable status. A cancellation that
|
||||
// preceded registration must not be erased by a fresh active scope.
|
||||
if ((await getRun(run.id))?.status !== "running") {
|
||||
throw Object.assign(new Error("Run stopped before adapter dispatch"), { code: "ADAPTER_RUN_CANCELLED" });
|
||||
}
|
||||
}
|
||||
const localAgentJwtScope =
|
||||
issueRef?.workMode === "skill_test"
|
||||
@@ -21671,8 +21676,9 @@ export function heartbeatService(
|
||||
}
|
||||
const guardedDispatch =
|
||||
await measureSandboxOperation("heartbeat.dispatch_resolved_interaction_continuation_with_atomic_gate", { operationIndex: 147 }, async () => (dispatchResolvedInteractionContinuationWithAtomicGate(
|
||||
(markDispatchStarted) =>
|
||||
adapter.execute({
|
||||
(markDispatchStarted) => {
|
||||
throwIfAdapterRunCancelled(run.id);
|
||||
return adapter.execute({
|
||||
runId: run.id,
|
||||
agent,
|
||||
runtime: runtimeForAdapter,
|
||||
@@ -21716,7 +21722,8 @@ export function heartbeatService(
|
||||
}, executionTarget?.kind === "remote" ? "remote" : "local")));
|
||||
},
|
||||
authToken: authToken ?? undefined,
|
||||
}),
|
||||
});
|
||||
},
|
||||
)));
|
||||
if (!guardedDispatch.dispatched) return;
|
||||
adapterResult = await measureSandboxOperation("heartbeat.guarded_dispatch.result_promise", { operationIndex: 150 }, async () => (guardedDispatch.resultPromise));
|
||||
@@ -26125,6 +26132,10 @@ export function heartbeatService(
|
||||
: {}),
|
||||
});
|
||||
|
||||
// Scope initialization may have raced the earlier lookup. Persist first,
|
||||
// then notify any scope now present; later scopes read this terminal state.
|
||||
if (run.runtimeMode !== "native") await cancelAdapterRunExecution(run.id);
|
||||
|
||||
await setWakeupStatus(run.wakeupRequestId, "cancelled", {
|
||||
finishedAt,
|
||||
error: reason,
|
||||
|
||||
Reference in new issue
Block a user