diff --git a/server/src/services/heartbeat/native-stop-pending.test.ts b/server/src/services/heartbeat/native-stop-pending.test.ts new file mode 100644 index 00000000000..258eafd144a --- /dev/null +++ b/server/src/services/heartbeat/native-stop-pending.test.ts @@ -0,0 +1,64 @@ +import type { Db } from "@paperclipai/db"; +import { heartbeatRuns } from "@paperclipai/db"; +import { describe, expect, it, vi } from "vitest"; +import { HttpError } from "../../errors.js"; +import { cancelNativeSession, NativeCancellationPendingRecoveryError } from "../native-runtime/index.js"; +import { createHeartbeatRunControl } from "./run-control.js"; +import { callbacks, runControlPostgresSupport, setupRunControlDatabase } from "./__tests__/run-control-fixtures.js"; + +vi.mock("../live-events.js", () => ({ publishLiveEvent: vi.fn() })); +vi.mock("../local-service-supervisor.js", async (importOriginal) => ({ + ...await importOriginal(), + terminateLocalService: vi.fn(async () => undefined), +})); +vi.mock("../native-runtime/index.js", async (importOriginal) => ({ + ...await importOriginal(), + cancelNativeSession: vi.fn(), +})); + +describe.skipIf(!runControlPostgresSupport.supported)("native Stop while the provider session is still starting", () => { + let db: Db; + const { fixture } = setupRunControlDatabase("paperclip-heartbeat-native-stop-pending", value => { db = value; }); + + async function startingNativeRun() { + const { company, agent } = await fixture(); + const [run] = await db.insert(heartbeatRuns).values({ + companyId: company.id, agentId: agent.id, status: "running", runtimeMode: "native", runtimeModeResolvedAt: new Date(), + }).returning(); + return run; + } + + // Staging: an operator Stop arrived while a remote native session took longer + // than the 30s startup wait to open. The Stop was durably recorded and later + // honored by the startup owner, but the request answered HTTP 500. + it.each([ + ["startup wait timed out", () => new NativeCancellationPendingRecoveryError()], + ["remote owner acknowledgement timed out", () => new Error("native_cancellation_pending_recovery")], + ])("answers a recorded but unacknowledged Stop (%s) with a pending conflict", async (_label, failure) => { + const run = await startingNativeRun(); + const deps = callbacks(db); + vi.mocked(cancelNativeSession).mockRejectedValueOnce(failure()); + const error = await createHeartbeatRunControl(db, deps) + .cancelRunInternal(run.id, "Cancelled by a board operator", { resultJson: { cancelledByActorType: "user", cancelledByUserId: "user-1" } }) + .catch((err: unknown) => err); + expect(error).toBeInstanceOf(HttpError); + expect(error).toMatchObject({ status: 409, details: { code: "native_stop_pending", runId: run.id } }); + // Pending is not a cancellation receipt: no terminal transition or release. + expect(deps.setRunStatusFromLive).not.toHaveBeenCalled(); + expect(deps.finalizeAgentStatus).not.toHaveBeenCalled(); + expect(deps.startNextQueuedRunForAgent).not.toHaveBeenCalled(); + const current = await deps.getRun(run.id); + expect(current?.status).toBe("running"); + // The Stop intent itself stays recorded for the startup owner to honor. + expect(current?.resultJson).toMatchObject({ cancellation: expect.any(Object), startupCancellation: expect.any(Object) }); + }); + + it("keeps other native Stop failures unchanged", async () => { + const run = await startingNativeRun(); + const deps = callbacks(db); + const failure = new Error("native_cancellation_intent_conflict"); + vi.mocked(cancelNativeSession).mockRejectedValueOnce(failure); + await expect(createHeartbeatRunControl(db, deps).cancelRunInternal(run.id)).rejects.toBe(failure); + expect(deps.setRunStatusFromLive).not.toHaveBeenCalled(); + }); +}); diff --git a/server/src/services/heartbeat/run-control.ts b/server/src/services/heartbeat/run-control.ts index 11f25b30457..9f609e40432 100644 --- a/server/src/services/heartbeat/run-control.ts +++ b/server/src/services/heartbeat/run-control.ts @@ -1462,13 +1462,25 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont if (!dependencies.verifyStartupCleanup) throw conflict("Native startup cleanup worker unavailable"); await dependencies.verifyStartupCleanup(startupCleanupRun); } else { - await cancelHeartbeatNativeRun({ - db, - runId: run.id, - reason, - runtimeMode: run.runtimeMode, - ...(options.cancellationRequestId ? { cancellationRequestId: options.cancellationRequestId } : {}), - }); + try { + await cancelHeartbeatNativeRun({ + db, + runId: run.id, + reason, + runtimeMode: run.runtimeMode, + ...(options.cancellationRequestId ? { cancellationRequestId: options.cancellationRequestId } : {}), + }); + } catch (error) { + // The durable Stop intent is recorded, but its owner (a native + // session still opening, or a remote controller) has not yet + // acknowledged it. The owner honors it later; answer honestly + // instead of reporting an internal failure. + if (error instanceof Error && error.message === "native_cancellation_pending_recovery") { + throw conflict("Stop is recorded; the native session is still starting and will stop when it opens.", + { code: "native_stop_pending", runId: run.id }); + } + throw error; + } } if (running) { await terminateHeartbeatRunProcess({ diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 39a403fe553..9f1898dea66 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -8374,6 +8374,53 @@ describe("native session cancellation", () => { } }); + it("acknowledges a timed-out Stop's durable intent when the late startup handle opens", async () => { + const root = await mkdtemp(join(tmpdir(), "native-late-startup-ack-")); + const previous = process.env.PAPERCLIP_RUNNER_STATE_DIR; + process.env.PAPERCLIP_RUNNER_STATE_DIR = root; + let open!: () => void, started!: () => void; + const opening = new Promise(resolve => { open = resolve; }); + const admitted = new Promise(resolve => { started = resolve; }); + const submitTurn = vi.fn(); + state.cancel.mockClear(); + state.execute.mockReset().mockImplementationOnce(async (options) => { + started(); await opening; + await options.onSession({ cancel: state.cancel }); + submitTurn(); + throw new Error("a stopped startup must not reach prompt submission"); + }); + const runResultJson: Record = {}; + const updates: Array<{ table: unknown; values: Record }> = []; + const outcome = executePaperclipNativeSession({ db: leaseDb(execution, {}, runResultJson, updates), execution, runnerInstanceId: "runner" }) + .catch(error => error); + const persistence = cancellationDb(); + try { + await admitted; + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + const stopping = cancelNativeSession(execution.binding.runId, "operator Stop", { db: persistence.db, scope: "run" }).catch(error => error); + await new Promise(resolve => setImmediate(resolve)); + await vi.advanceTimersByTimeAsync(30_000); + expect(await stopping).toMatchObject({ name: "NativeCancellationPendingRecoveryError" }); + vi.useRealTimers(); + // The startup owner reads the same durable run row the Stop caller wrote. + Object.assign(runResultJson, persistence.getResultJson()); + expect(runResultJson.nativeCancellation).toMatchObject({ dispatchState: "pending", scope: "run" }); + open(); await outcome; + expect(state.cancel).toHaveBeenCalledOnce(); + expect(submitTurn).not.toHaveBeenCalled(); + const acknowledgement = updates.find(update => update.table === heartbeatRuns && + (update.values.resultJson as Record | undefined)?.nativeCancellation?.dispatchState === "acknowledged"); + expect(acknowledgement?.values.resultJson).toMatchObject({ + nativeCancellation: { intentId: (runResultJson.nativeCancellation as Record).intentId, dispatched: true }, + }); + } finally { + vi.useRealTimers(); open(); await outcome; + if (previous === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; + else process.env.PAPERCLIP_RUNNER_STATE_DIR = previous; + await rm(root, { recursive: true, force: true }); + } + }); + it("routes control-plane cancellation to the active normalized session and removes the handle", async () => { const running = executePaperclipNativeSession({ db: leaseDb(), diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index da9b82fbb41..a70335eb7b0 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -10353,6 +10353,24 @@ async function executePaperclipNativeSessionWithinScope( await startup.cancellationSettled; // If startup exceeded the caller's deadline, its late handle // must still be cancelled instead of escaping the Stop fence. + // The caller left its durable run Stop pending; acknowledge it + // against this exact published handle so the run settles as an + // operator cancellation rather than a provider failure. + try { + const [stopRun] = await input.db.select({ resultJson: heartbeatRuns.resultJson }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, input.execution.binding.runId), + eq(heartbeatRuns.companyId, input.execution.binding.companyId), + )).limit(1); + const pendingStop = record(record(stopRun?.resultJson).nativeCancellation); + if (pendingStop.scope === "run" && pendingStop.dispatchState === "pending") { + await cancelNativeSession(input.execution.binding.runId, "Stop requested during native startup", { db: input.db, scope: "run" }); + } + } catch (error) { + // Best effort: the in-memory Stop below still fences this + // handle; the intent stays pending for cancellation recovery. + logger.warn({ runId: input.execution.binding.runId, err: error }, + "Late native startup Stop could not acknowledge its durable intent"); + } await cancelNativeSession(input.execution.binding.runId, "Stop requested during native startup"); throw new Error("native_finalization_missing: session returned no semantic result"); }