mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 20:46:28 +01:00
Stop legacy sandbox execution before completing cancellation
Bind remote CLI and ACP processes to host-owned cancellation scopes, wait for teardown and final saves, and distinguish deliberate bridge shutdown from transport loss. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
8cee86df73
commit
066b4f0160
9 files changed
+544
-83
No files matched your search
@@ -5,6 +5,8 @@ import { agents, companies, createDb, heartbeatRuns, startEmbeddedPostgresTestDa
|
||||
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 { 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";
|
||||
|
||||
@@ -66,6 +68,58 @@ describe("heartbeat process identity persistence", () => {
|
||||
}
|
||||
}, 30_000);
|
||||
|
||||
it("cancels the remote adapter and waits for teardown before acknowledging", 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);
|
||||
vi.spyOn(executionTargets, "prepareGitHubOperationLaunchers").mockImplementation(async (input) => input.env);
|
||||
vi.spyOn(executionTargets, "cleanupGitHubOperationLaunchers").mockResolvedValue(undefined);
|
||||
let ready!: () => void, stopped!: () => void, release!: () => void;
|
||||
const started = new Promise<void>((resolve) => { ready = resolve; });
|
||||
const interrupted = new Promise<void>((resolve) => { stopped = resolve; });
|
||||
const teardown = new Promise<void>((resolve) => { release = resolve; });
|
||||
const heartbeat = heartbeatService(db);
|
||||
vi.spyOn(adapters, "getServerAdapter").mockReturnValue({
|
||||
supportsLocalAgentJwt: false,
|
||||
execute: async (input) => {
|
||||
expect(hasAdapterRunCancellation(input.runId)).toBe(true);
|
||||
const cleanup = await bindAdapterRunStop(input.runId, async () => {
|
||||
expect((await heartbeat.getRun(input.runId))?.status).toBe("cancelled");
|
||||
stopped();
|
||||
});
|
||||
ready();
|
||||
await interrupted;
|
||||
await teardown;
|
||||
await cleanup();
|
||||
return { exitCode: 143, signal: "SIGTERM", timedOut: false };
|
||||
},
|
||||
} as ReturnType<typeof adapters.getServerAdapter>);
|
||||
let pending: Promise<unknown> | undefined;
|
||||
try {
|
||||
const queued = await heartbeat.invoke(agentId, "on_demand", {}, "manual");
|
||||
expect(queued).not.toBeNull();
|
||||
await started;
|
||||
let acknowledged = false;
|
||||
pending = heartbeat.cancelRun(queued!.id).then((result) => { acknowledged = true; return result; });
|
||||
await interrupted;
|
||||
expect(acknowledged).toBe(false);
|
||||
release();
|
||||
await pending;
|
||||
expect((await heartbeat.getRun(queued!.id))?.status).toBe("cancelled");
|
||||
expect(hasAdapterRunCancellation(queued!.id)).toBe(false);
|
||||
} finally {
|
||||
stopped(); release();
|
||||
await pending;
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
}
|
||||
}, 30_000);
|
||||
|
||||
it("uses the remote marker even when its PID exists on the host", async () => {
|
||||
const run = await running();
|
||||
const host = vi.spyOn(processes, "readProcessStartedAt");
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { beginAdapterRunCancellation, cancelAdapterRunExecution, finishAdapterRunCancellation, hasAdapterRunCancellation } 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";
|
||||
@@ -13410,6 +13411,7 @@ export function heartbeatService(
|
||||
},
|
||||
);
|
||||
if (!interruptedStatus.updated || !interruptedStatus.run) continue;
|
||||
if (run.runtimeMode !== "native") await cancelAdapterRunExecution(run.id);
|
||||
let interrupted = interruptedStatus.run;
|
||||
await setWakeupStatus(run.wakeupRequestId, "cancelled", {
|
||||
finishedAt: now,
|
||||
@@ -17990,10 +17992,14 @@ export function heartbeatService(
|
||||
|
||||
// The diagnostic path is opt-in and persists bounded batches after execution.
|
||||
async function executeRun(runId: string, runOptions: Parameters<typeof executeRunMeasured>[1] = {}) {
|
||||
return runWithSandboxPerformanceTrace({ runId, onBatch: async (batch) => {
|
||||
const observed = await getRun(runId);
|
||||
if (observed) await appendRunEvent(observed, { eventType: "sandbox.performance.batch", stream: "system", level: "info", payload: batch });
|
||||
} }, () => executeRunMeasured(runId, runOptions));
|
||||
try {
|
||||
return await runWithSandboxPerformanceTrace({ runId, onBatch: async (batch) => {
|
||||
const observed = await getRun(runId);
|
||||
if (observed) await appendRunEvent(observed, { eventType: "sandbox.performance.batch", stream: "system", level: "info", payload: batch });
|
||||
} }, () => executeRunMeasured(runId, runOptions));
|
||||
} finally {
|
||||
finishAdapterRunCancellation(runId);
|
||||
}
|
||||
}
|
||||
|
||||
async function executeRunMeasured(
|
||||
@@ -21142,6 +21148,10 @@ export function heartbeatService(
|
||||
.where(eq(heartbeatRuns.id, run.id))));
|
||||
}
|
||||
setSandboxPerformanceRunAttributes({ runtime: nativeRuntimeResolution.kind });
|
||||
if (nativeRuntimeResolution.kind === "legacy"
|
||||
&& executionTarget?.kind === "remote" && executionTarget.transport === "sandbox") {
|
||||
beginAdapterRunCancellation(run.id);
|
||||
}
|
||||
const localAgentJwtScope =
|
||||
issueRef?.workMode === "skill_test"
|
||||
? { kind: "skill_test" as const, issueId: issueRef.id }
|
||||
@@ -26034,8 +26044,13 @@ export function heartbeatService(
|
||||
!CANCELLABLE_HEARTBEAT_RUN_STATUSES.includes(
|
||||
run.status as (typeof CANCELLABLE_HEARTBEAT_RUN_STATUSES)[number],
|
||||
)
|
||||
)
|
||||
) {
|
||||
if (run.status === "cancelled" && hasAdapterRunCancellation(run.id)) {
|
||||
await cancelAdapterRunExecution(run.id);
|
||||
return await getRun(run.id);
|
||||
}
|
||||
return run;
|
||||
}
|
||||
const agent = await getAgent(run.agentId);
|
||||
const errorCode = options.errorCode ?? "cancelled";
|
||||
const resultJson = agent
|
||||
@@ -26049,6 +26064,14 @@ export function heartbeatService(
|
||||
}
|
||||
: options.resultJson;
|
||||
|
||||
if (run.runtimeMode !== "native" && hasAdapterRunCancellation(run.id)) {
|
||||
await setRunStatus(run.id, "cancelled", {
|
||||
finishedAt: new Date(), error: reason, errorCode,
|
||||
...(resultJson ? { resultJson } : {}),
|
||||
});
|
||||
await cancelAdapterRunExecution(run.id);
|
||||
}
|
||||
|
||||
const running = runningProcesses.get(run.id);
|
||||
try {
|
||||
await cancelHeartbeatNativeRun({
|
||||
@@ -26186,6 +26209,7 @@ export function heartbeatService(
|
||||
graceMs: Math.max(1, running.graceSec) * 1000,
|
||||
});
|
||||
}
|
||||
if (run.runtimeMode !== "native") await cancelAdapterRunExecution(run.id);
|
||||
runningProcesses.delete(run.id);
|
||||
await releaseIssueExecutionAndPromote(run);
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user