fix(native): bind operator Stop to caller cancellation intent

Reserve an optional board request UUID under the run lock and retain it
through default Stop joins and native dispatch. Reject prior or competing
intents and require the same actor for idempotent retries.

Bind the Copilot denial fixture to its exact request and audited intent,
with versioned causal receipts and negative controls for earlier Stop.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

server/src/__tests__/heartbeat-native-runner-cancellation.test.ts

@@ -41,6 +41,13 @@ describe("native heartbeat cancellation authority", () => {
     );
   });
 
+  it("passes the caller identity into the audited native boundary", async () => {
+    const cancel = vi.fn(async () => ({ decision: {}, auditId: "audit" }));
+    const db = {} as Db, cancellationRequestId = "11111111-1111-4111-8111-111111111111";
+    await cancelHeartbeatNativeRun({ db, runId: "run", reason: "Stop", runtimeMode: "native", cancellationRequestId, cancel });
+    expect(cancel).toHaveBeenCalledWith("run", "Stop", { db, scope: "run", cancellationRequestId });
+  });
+
   it("fails closed when a native cancellation lacks its decision audit", async () => {
     const cancel = vi.fn(async () => ({
       decision: null,

server/src/__tests__/native-cancellation-request.integration.test.ts

@@ -0,0 +1,54 @@
+import { randomUUID } from "node:crypto";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { eq, sql } from "drizzle-orm";
+import { agents, companies, createDb, heartbeatRuns } from "@paperclipai/db";
+import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
+import { cancellationIntentId, claimCancellationRequest, startupCancellationFence } from "../services/native-runtime/native-cancellation-request.js";
+
+describe("atomic caller cancellation request ownership", () => {
+  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+  let db: ReturnType<typeof createDb>;
+  beforeAll(async () => { database = await startEmbeddedPostgresTestDatabase("caller-stop-"); db = createDb(database.connectionString); }, 90_000);
+  afterAll(async () => { await database?.cleanup(); });
+  async function fixture() {
+    const companyId = randomUUID(), agentId = randomUUID(), runId = randomUUID();
+    await db.insert(companies).values({ id: companyId, name: "Stop claims", issuePrefix: randomUUID().slice(0, 8) });
+    await db.insert(agents).values({ id: agentId, companyId, name: "Stop target" });
+    await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "on_demand", status: "running", runtimeMode: "native" });
+    return { companyId, runId };
+  }
+  const defaultFence = (runId: string) => db.update(heartbeatRuns).set({ resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || jsonb_build_object('startupCancellation', ${startupCancellationFence(new Date().toISOString())})` }).where(eq(heartbeatRuns.id, runId));
+  it("allows exactly one of two concurrent identities and idempotent same-ID retries", async () => {
+    const { companyId, runId } = await fixture(), a = randomUUID(), b = randomUUID();
+    const results = await Promise.allSettled([claimCancellationRequest(db, runId, companyId, a, "board-user"), claimCancellationRequest(db, runId, companyId, b, "board-user")]);
+    expect(results.filter(r => r.status === "fulfilled")).toHaveLength(1);
+    const winner = results[0].status === "fulfilled" ? a : b;
+    await expect(claimCancellationRequest(db, runId, companyId, winner, "other-board-user")).rejects.toMatchObject({ status: 409 });
+    const copies = await Promise.all([claimCancellationRequest(db, runId, companyId, winner, "board-user"), claimCancellationRequest(db, runId, companyId, winner, "board-user")]);
+    expect(copies[0].resultJson).toEqual(copies[1].resultJson);
+    await expect(claimCancellationRequest(db, runId, randomUUID(), winner, "board-user")).rejects.toMatchObject({ status: 409 });
+  });
+  it("does not overwrite a caller claim when default Stop races after it", async () => {
+    const { companyId, runId } = await fixture(), id = randomUUID();
+    const claimed = await claimCancellationRequest(db, runId, companyId, id, "board-user");
+    await Promise.all([defaultFence(runId), claimCancellationRequest(db, runId, companyId, id, "board-user")]);
+    const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
+    expect(run.resultJson).toEqual(claimed.resultJson);
+  });
+  it("rejects a default Stop that won first, without replacing its marker", async () => {
+    const { companyId, runId } = await fixture(); await defaultFence(runId);
+    const [before] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
+    await expect(claimCancellationRequest(db, runId, companyId, randomUUID(), "board-user")).rejects.toMatchObject({ status: 409 });
+    const [after] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
+    expect(after.resultJson).toEqual(before.resultJson);
+  });
+  it("accepts only the original acknowledged identity on a terminal run", async () => {
+    const { companyId, runId } = await fixture(), id = randomUUID();
+    const run = await claimCancellationRequest(db, runId, companyId, id, "board-user");
+    await db.update(heartbeatRuns).set({ status: "cancelled", resultJson: { ...run.resultJson, nativeCancellation: { schema: "paperclip.native-cancellation.v1", intentId: cancellationIntentId(id), runId, companyId, scope: "run", dispatched: true, dispatchState: "acknowledged", intentAuditId: randomUUID(), acknowledgementAuditId: randomUUID() } } }).where(eq(heartbeatRuns.id, runId));
+    await expect(claimCancellationRequest(db, runId, companyId, id, "board-user")).resolves.toMatchObject({ status: "cancelled" });
+    await expect(claimCancellationRequest(db, runId, companyId, randomUUID(), "board-user")).rejects.toMatchObject({ status: 409 });
+    await db.update(heartbeatRuns).set({ resultJson: { ...run.resultJson, nativeCancellation: { intentId: cancellationIntentId(id) } } }).where(eq(heartbeatRuns.id, runId));
+    await expect(claimCancellationRequest(db, runId, companyId, id, "board-user")).rejects.toMatchObject({ status: 409 });
+  });
+});

server/src/routes/agents.ts

@@ -1,3 +1,4 @@
+import { cancellationRequestId } from "../services/native-runtime/native-cancellation-request.js";
 import { agentFileStore, agentFileTokenFromHash } from "../services/agent-file-store.js";
 import { pipeline } from "node:stream/promises";
 import { resolveAgentAppearance, agentAvatarUrl } from "@paperclipai/shared";
@@ -6958,10 +6959,12 @@ export function agentRoutes(
     const runId = readHeartbeatRunId(req);
     const existing = await getAccessibleResource(req, res, heartbeat.getRun(runId), "Heartbeat run not found");
     if (!existing) return;
+    const requestId = cancellationRequestId(req.body?.cancellationRequestId);
     // Stamp the cancellation as operator-initiated (this route is board-only).
     // Recovery reads this to stand down instead of classifying the cancelled
     // run as agent stranding and re-waking the agent the operator just stopped.
     const run = await heartbeat.cancelRun(runId, "Cancelled by a board operator", {
+      ...(requestId ? { cancellationRequestId: requestId, cancellationRequestedByUserId: req.actor.userId ?? null } : {}),
       resultJson: {
         cancelledByActorType: "user",
         cancelledByUserId: req.actor.userId ?? null,

server/src/services/heartbeat.ts

@@ -1,3 +1,4 @@
+import { claimCancellationRequest, startupCancellationFence } from "./native-runtime/native-cancellation-request.js";
 import { CHAT_COMPLETION_WAKE_REASON, prepareChatCompletionTurn, chatCompletionInstruction, isCompletedOnboardingHandoffWake } from "./chat-completion-delivery.js";
 import { isAgentDirectoryCopy } from "./agent-directory-working-copies.js";
 
@@ -9358,10 +9359,11 @@ export async function cancelHeartbeatNativeRun(input: {
   runId: string;
   reason: string;
   runtimeMode: string | null;
+  cancellationRequestId?: string;
   cancel?: (
     runId: string,
     reason: string,
-    options: { db: Db; scope: "run" },
+    options: { db: Db; scope: "run"; cancellationRequestId?: string },
   ) => Promise<{ decision: unknown | null; auditId: string | null }>;
 }) {
   if (input.runtimeMode !== "native") {
@@ -9371,10 +9373,12 @@ export async function cancelHeartbeatNativeRun(input: {
     ? await input.cancel(input.runId, input.reason, {
         db: input.db,
         scope: "run",
+        ...(input.cancellationRequestId ? { cancellationRequestId: input.cancellationRequestId } : {}),
       })
     : await cancelNativeSession(input.runId, input.reason, {
         db: input.db,
         scope: "run",
+        ...(input.cancellationRequestId ? { cancellationRequestId: input.cancellationRequestId } : {}),
       });
   if (!cancellation.decision || !cancellation.auditId) {
     throw new Error("native_cancellation_outcome_not_audited");
@@ -29108,6 +29112,9 @@ export function heartbeatService(
   }
 
   type CancelRunOptions = {
+    /** Optional board request identity, atomically reserved for native Stop. */
+    cancellationRequestId?: string;
+    cancellationRequestedByUserId?: string | null;
     errorCode?: string;
     resultJson?: Record<string, unknown>;
     eventMessage?: string;
@@ -29136,6 +29143,9 @@ export function heartbeatService(
   ) {
     let run = await getRun(runId);
     if (!run) throw notFound("Heartbeat run not found");
+    if (options.cancellationRequestId) {
+      run = await claimCancellationRequest(db, runId, run.companyId, options.cancellationRequestId, options.cancellationRequestedByUserId ?? null);
+    }
     const pendingNativeRetry =
       run.runtimeMode === "native" && run.status === "failed"
         ? await db
@@ -29181,20 +29191,14 @@ export function heartbeatService(
     // Established legacy processes must still be stopped if the database is
     // unavailable. Only native or not-yet-dispatched preparation needs this
     // additional durable fence before its existing cancellation path.
-    if (run.runtimeMode === "native" || (!run.runtimeModeResolvedAt && !running && !control)) {
+    if (!options.cancellationRequestId && (run.runtimeMode === "native" || (!run.runtimeModeResolvedAt && !running && !control))) {
       const [fenced] = await db.update(heartbeatRuns).set({
         // Record handoff intent before native cancellation can finalize and release
         // the run. Only its audited stop acknowledgement suppresses recovery.
         resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) ||
           ${JSON.stringify(options.errorCode === "issue_reassigned" && options.resultJson?.reassignmentStopConfirmed === true
             ? { reassignmentStopRequested: true } : {})}::jsonb ||
-          jsonb_build_object('startupCancellation', jsonb_build_object(
-            'requestedAt', ${new Date().toISOString()}::text,
-            'beforeNativeSelection', ${heartbeatRuns.runtimeMode} = 'legacy'
-              and ${heartbeatRuns.runtimeModeResolvedAt} is null
-              and ${heartbeatRuns.executionStage} = 'preparing'
-              and coalesce(${heartbeatRuns.runnerProfileJson}->'adapterDispatch'->>'adapterType' = 'paperclip_runner', false)
-          ))`,
+          jsonb_build_object('startupCancellation', ${startupCancellationFence(new Date().toISOString())})`,
       }).where(and(eq(heartbeatRuns.id, runId), inArray(heartbeatRuns.status,
         pendingNativeRetry ? [...CANCELLABLE_HEARTBEAT_RUN_STATUSES, "failed"] : [...CANCELLABLE_HEARTBEAT_RUN_STATUSES],
       ))).returning();
@@ -29270,6 +29274,7 @@ export function heartbeatService(
               runId: run.id,
               reason,
               runtimeMode: run.runtimeMode,
+              ...(options.cancellationRequestId ? { cancellationRequestId: options.cancellationRequestId } : {}),
             });
             if (running) {
               await terminateHeartbeatRunProcess({

server/src/services/native-runtime/native-cancellation-request.test.ts

@@ -0,0 +1,23 @@
+import { describe, expect, it } from "vitest";
+import { assertCancellationRequest, cancellationIntentId, cancellationRequestId } from "./native-cancellation-request.js";
+const id = "11111111-1111-4111-8111-111111111111";
+describe("caller-bound native cancellation", () => {
+  it.each([null, "", " x", {}, [], 1, `${id} `, "not-a-uuid"])("rejects malformed request %j", value => {
+    expect(() => cancellationRequestId(value)).toThrow("Invalid cancellationRequestId");
+  });
+  it("preserves the absent option and normalizes a valid UUID", () => {
+    expect(cancellationRequestId(undefined)).toBeUndefined();
+    expect(cancellationRequestId("AAAAAAAA-AAAA-4AAA-8AAA-AAAAAAAAAAAA")).toBe("aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa");
+  });
+  it.each([{ startupCancellation: {} }, { startupCancellation: { requestedAt: "earlier" } },
+    { startupCancellation: { cancellationRequestId: "other" } }, { nativeCancellation: { intentId: "other" } }])("rejects prior or racing intent %j", result => {
+    expect(() => assertCancellationRequest(result, id, true)).toThrow("earlier Stop");
+  });
+  it("requires a reservation at the native locked dispatch gate", () => {
+    expect(() => assertCancellationRequest({}, id)).toThrow("earlier Stop");
+    expect(() => assertCancellationRequest({}, id, true)).not.toThrow();
+    const startupCancellation = { cancellationRequestId: id };
+    expect(() => assertCancellationRequest({ startupCancellation }, id)).not.toThrow();
+    expect(() => assertCancellationRequest({ startupCancellation, nativeCancellation: { intentId: cancellationIntentId(id) } }, id)).not.toThrow();
+  });
+});

server/src/services/native-runtime/native-cancellation-request.ts

@@ -0,0 +1,71 @@
+import { and, eq, sql } from "drizzle-orm";
+import { heartbeatRuns, type Db } from "@paperclipai/db";
+import { badRequest, conflict } from "../../errors.js";
+
+const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
+const record = (value: unknown): Record<string, unknown> => value !== null && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : {};
+export function cancellationRequestId(value: unknown): string | undefined {
+  if (value === undefined) return undefined;
+  if (typeof value !== "string" || !UUID.test(value)) throw badRequest("Invalid cancellationRequestId");
+  return value.toLowerCase();
+}
+export const cancellationIntentId = (requestId: string) => `native-cancellation:${requestId}`;
+
+/** Run only while holding this company's run row lock. An earlier uncorrelated
+ * startup fence is also a conflicting Stop, even before native intent exists. */
+export function assertCancellationRequest(result: unknown, requestId: string, allowUnclaimed = false) {
+  const value = record(result), startup = record(value.startupCancellation), native = record(value.nativeCancellation);
+  if ((Object.hasOwn(value, "startupCancellation") && startup.cancellationRequestId !== requestId)
+    || (Object.hasOwn(value, "nativeCancellation") && native.intentId !== cancellationIntentId(requestId))
+    || (!allowUnclaimed && startup.cancellationRequestId !== requestId)) {
+    throw conflict("Cancellation request conflicts with an earlier Stop");
+  }
+}
+
+/** Reserve the caller's identity before handoff or dispatch. No schema change:
+ * the existing startup fence and native intent carry the correlation. */
+export async function claimCancellationRequest(db: Db, runId: string, companyId: string, requestId: string, userId: string | null) {
+  if (cancellationRequestId(requestId) !== requestId) throw badRequest("Invalid cancellationRequestId");
+  if (userId !== null && (typeof userId !== "string" || !userId)) throw badRequest("Invalid cancellation request actor");
+  return db.transaction(async tx => {
+    const run = await tx.select().from(heartbeatRuns)
+      .where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, companyId)))
+      .for("update").limit(1).then(rows => rows[0]);
+    if (!run || run.runtimeMode !== "native") throw conflict("Correlated Stop requires a native run");
+    assertCancellationRequest(run.resultJson, requestId, true);
+    const result = record(run.resultJson);
+    if (record(result.startupCancellation).cancellationRequestId === requestId) {
+      const actor = record(record(result.startupCancellation).requestedBy);
+      if (actor.type !== "board" || actor.userId !== userId) throw conflict("Cancellation request belongs to another actor");
+      if (!["queued", "running", "scheduled_retry"].includes(run.status)) {
+        const native = record(result.nativeCancellation);
+        if (run.status !== "cancelled" || native.schema !== "paperclip.native-cancellation.v1"
+          || native.intentId !== cancellationIntentId(requestId) || native.runId !== runId || native.companyId !== companyId
+          || native.scope !== "run" || native.dispatchState !== "acknowledged" || typeof native.dispatched !== "boolean"
+          || typeof native.intentAuditId !== "string" || !native.intentAuditId
+          || typeof native.acknowledgementAuditId !== "string" || !native.acknowledgementAuditId
+          || native.intentAuditId === native.acknowledgementAuditId) throw conflict("Correlated Stop run is already terminal without acknowledgement");
+      }
+      return run;
+    }
+    if (!["queued", "running", "scheduled_retry"].includes(run.status)) throw conflict("Correlated Stop run is already terminal");
+    const [claimed] = await tx.update(heartbeatRuns).set({ resultJson: { ...result,
+      startupCancellation: { requestedAt: new Date().toISOString(), beforeNativeSelection: false, cancellationRequestId: requestId, requestedBy: { type: "board", userId } },
+    } }).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, companyId))).returning();
+    if (!claimed) throw conflict("Correlated Stop run changed");
+    return claimed;
+  });
+}
+
+/** Used by the default Stop path too: never erase a correlated reservation. */
+export function startupCancellationFence(requestedAt: string) {
+  return sql`CASE
+    WHEN ${heartbeatRuns.resultJson}->'startupCancellation'->>'cancellationRequestId' IS NOT NULL
+      THEN ${heartbeatRuns.resultJson}->'startupCancellation'
+    ELSE jsonb_build_object('requestedAt', ${requestedAt}::text,
+      'beforeNativeSelection', ${heartbeatRuns.runtimeMode} = 'legacy'
+        and ${heartbeatRuns.runtimeModeResolvedAt} is null
+        and ${heartbeatRuns.executionStage} = 'preparing'
+        and coalesce(${heartbeatRuns.runnerProfileJson}->'adapterDispatch'->>'adapterType' = 'paperclip_runner', false))
+    END`;
+}

server/src/services/native-runtime/native-session-executor.test.ts

@@ -4919,6 +4919,7 @@ function cancellationDb(options?: {
   } | null;
   failResultJsonUpdateAt?: number;
   ownershipHeld?: boolean;
+  resultJson?: Record<string, unknown>;
 }) {
   const initialRun = {
     id: execution.binding.runId,
@@ -4938,6 +4939,7 @@ function cancellationDb(options?: {
   };
   let currentResultJson: Record<string, unknown> = {
     durableReceipt: { operationId: "operation-1" },
+    ...options?.resultJson,
   };
   const issue = {
     status: "in_progress",
@@ -5505,6 +5507,34 @@ describe("native session cancellation", () => {
     expect(state.publishActivity).toHaveBeenCalledTimes(2);
   });
 
+  it("uses the reserved caller UUID and permits same-intent and default retries", async () => {
+    const id = "11111111-1111-4111-8111-111111111111";
+    const persistence = cancellationDb({ resultJson: { startupCancellation: { cancellationRequestId: id } } });
+    await cancelNativeSession(execution.binding.runId, "operator Stop", { db: persistence.db, scope: "run", cancellationRequestId: id });
+    expect(persistence.getResultJson().nativeCancellation).toMatchObject({ intentId: `native-cancellation:${id}` });
+    const writes = persistence.getResultJsonUpdateCount();
+    await cancelNativeSession(execution.binding.runId, "same request", { db: persistence.db, scope: "run", cancellationRequestId: id });
+    await cancelNativeSession(execution.binding.runId, "default retry", { db: persistence.db, scope: "run" });
+    expect(persistence.getResultJsonUpdateCount()).toBe(writes);
+    await expect(cancelNativeSession(execution.binding.runId, "foreign caller", {
+      db: persistence.db, scope: "run", cancellationRequestId: "22222222-2222-4222-8222-222222222222",
+    })).rejects.toMatchObject({ status: 409 });
+    expect(persistence.getResultJsonUpdateCount()).toBe(writes);
+  });
+
+  it.each([{}, { startupCancellation: { requestedAt: "prior" } },
+    { startupCancellation: { cancellationRequestId: "other" } },
+    { startupCancellation: { cancellationRequestId: "11111111-1111-4111-8111-111111111111" }, nativeCancellation: { intentId: "foreign" } },
+  ])("rejects missing or raced caller ownership under the intent lock: %j", async resultJson => {
+    const persistence = cancellationDb({ resultJson });
+    await expect(cancelNativeSession(execution.binding.runId, "operator Stop", {
+      db: persistence.db, scope: "run", cancellationRequestId: "11111111-1111-4111-8111-111111111111",
+    })).rejects.toMatchObject({ status: 409 });
+    expect(persistence.getForUpdateCount()).toBe(1);
+    expect(persistence.updates).toEqual([]);
+    expect(state.cancel).not.toHaveBeenCalled();
+  });
+
   it("recovers a post-dispatch persistence failure without cancelling the provider twice", async () => {
     const persistence = cancellationDb({ failResultJsonUpdateAt: 2 });
     const running = executePaperclipNativeSession({

server/src/services/native-runtime/native-session-executor.ts

@@ -1,3 +1,4 @@
+import { assertCancellationRequest, cancellationIntentId as callerCancellationIntentId, cancellationRequestId } from "./native-cancellation-request.js";
 import { readNativeCursorPlanWait } from "./native-cursor-plan-wait.js";
 import { resolveAcpxQualification } from "./acpx-qualification.js";
 import { readLocalAiCredentialFile } from "../local-ai-credential-file.js";
@@ -6527,6 +6528,7 @@ export function cancelNativeSession(
     db: Db;
     scope?: "turn" | "run" | "issue";
     replacementAccepted?: boolean;
+    cancellationRequestId?: string;
   },
 ): Promise<{
   dispatched: boolean;
@@ -6541,6 +6543,7 @@ export async function cancelNativeSession(
     db: Db;
     scope?: "turn" | "run" | "issue";
     replacementAccepted?: boolean;
+    cancellationRequestId?: string;
   },
 ): Promise<
   | boolean
@@ -6683,6 +6686,11 @@ export async function cancelNativeSession(
         throw new Error("native_cancellation_coordinator_missing");
 
       const resultJson = record(lockedRun.resultJson);
+      // A default Stop joins an already reserved caller intent; it cannot
+      // replace that intent while the originating request is still dispatching.
+      const requestedId = options.cancellationRequestId
+        ?? cancellationRequestId(record(resultJson.startupCancellation).cancellationRequestId);
+      if (requestedId) assertCancellationRequest(resultJson, requestedId);
       const existing = record(resultJson.nativeCancellation);
       const existingIntentId =
         typeof existing.intentId === "string" && existing.intentId.length > 0
@@ -6723,7 +6731,9 @@ export async function cancelNativeSession(
         };
       }
 
-      const intentId = `native-cancellation:${randomUUID()}`;
+      const intentId = requestedId
+        ? callerCancellationIntentId(requestedId)
+        : `native-cancellation:${randomUUID()}`;
       const activity = await persistActivity(tx as unknown as Db, {
         companyId: cancellationContext.companyId,
         actorType: "system",