From 8b62b64d690c9b2b5c8a242ed505d4f29f023c1f Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 19 Sep 2026 10:03:01 -0500 Subject: [PATCH] Keep native reassignment stops out of recovery and retries Co-Authored-By: Paperclip --- .../docs/capability-semantic-tools.md | 4 +++- .../wake-queue/adapters/postgres.test.ts | 24 +++++++++++++++++++ .../modules/wake-queue/adapters/postgres.ts | 9 +++---- .../services/acknowledged-native-stop.test.ts | 15 +++++++++++- .../src/services/acknowledged-native-stop.ts | 21 ++++++++++++---- server/src/services/heartbeat.ts | 8 +++++-- .../native-session-executor.test.ts | 6 +++-- .../native-runtime/native-session-executor.ts | 4 ++-- 8 files changed, 75 insertions(+), 16 deletions(-) diff --git a/packages/paperclip-runner/docs/capability-semantic-tools.md b/packages/paperclip-runner/docs/capability-semantic-tools.md index ee5ae11d7be..a09753d159d 100644 --- a/packages/paperclip-runner/docs/capability-semantic-tools.md +++ b/packages/paperclip-runner/docs/capability-semantic-tools.md @@ -108,7 +108,9 @@ documents, dependencies, scope, priority, and history. The old execution and any live goal must stop before ownership changes. Active work returns to todo; backlog and blocked work retain their status and are not -started. Assignment changes advance the observed version, including ordinary +started. An audited reassignment stop does not create a recovery blocker or +retry the outgoing run when it ends without a semantic completion result. +Assignment changes advance the observed version, including ordinary API changes, to reject stale handoffs. A durable audit receipt makes retries idempotent across caller runs; retrying after failed dispatch repairs the wake with a stable key and an owner/status/version guard. diff --git a/server/src/modules/wake-queue/adapters/postgres.test.ts b/server/src/modules/wake-queue/adapters/postgres.test.ts index 91a68b6b878..812cdec546b 100644 --- a/server/src/modules/wake-queue/adapters/postgres.test.ts +++ b/server/src/modules/wake-queue/adapters/postgres.test.ts @@ -318,6 +318,30 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { }); } + it("releases an acknowledged native handoff without blocking or restarting the old owner", async () => { + const companyId = await seedCompany(); + const agentId = await seedAgent({ companyId }); + const issueId = await seedIssue({ companyId, assigneeAgentId: agentId, status: "in_progress" }); + const runId = await seedRun({ companyId, agentId, status: "cancelled", contextSnapshot: { issueId } }); + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issueId, resultJson: { + reassignmentStopRequested: true, + nativeCancellation: { schema: "paperclip.native-cancellation.v1", runId, companyId, issueId, + scope: "run", reasonCode: "cancellation_run_only", dispatchState: "acknowledged", dispatched: true, + intentAuditId: randomUUID(), acknowledgementAuditId: randomUUID() }, + } }).where(eq(heartbeatRuns.id, runId)); + await db.update(issues).set({ executionRunId: runId, checkoutRunId: runId }).where(eq(issues.id, issueId)); + const before = (await db.select().from(issues).where(eq(issues.id, issueId)))[0]; + const wakeId = await seedDeferredWake({ companyId, agentId, issueId }); + const adapter = createPostgresWakeQueueAdapter(db, stubDeps); + const result = await adapter.withIssueExecutionLock({ companyId, runId, now: new Date() }, async () => { throw new Error("must not restart the outgoing owner"); }); + expect(result).toMatchObject({ outcome: { kind: "released" }, postCommitEffects: [] }); + expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]).toMatchObject({ + status: "in_progress", statusVersion: before.statusVersion, assigneeAgentId: agentId, executionRunId: null, checkoutRunId: null, + }); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wakeId)))[0].status).toBe("deferred_issue_execution"); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId))).toHaveLength(0); + }); + it.each(["in_progress", "blocked"])("preserves recovery ownership and queued messages when a native task fails from %s", async (status) => { const companyId = await seedCompany(); const agentId = await seedAgent({ companyId }); diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index 24fd0a21467..c43f0c43936 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -1,4 +1,4 @@ -import { isAcknowledgedNativeStop } from "../../../services/acknowledged-native-stop.js"; +import { isAcknowledgedNativeReassignmentStop, isAcknowledgedNativeStop } from "../../../services/acknowledged-native-stop.js"; import { instanceSettingsService } from "../../../services/instance-settings.js"; import { currentConversationCommentCondition } from "../../../services/agent-conversations.js"; import { getExecutionBlocker } from "../../../services/execution-blocker.js"; @@ -718,7 +718,7 @@ async function recordNativeTerminalRecoveryIfNeeded(tx: Db, run: HeartbeatRunRow ["failed", "timed_out", "interrupted", "cancelled"].includes(run.status) && issue.assigneeAgentId === run.agentId && !["done", "cancelled"].includes(issue.status); - if (!applies || isAcknowledgedNativeStop(run)) return false; + if (!applies || isAcknowledgedNativeStop(run) || isAcknowledgedNativeReassignmentStop(run)) return false; const existing = await tx .select({ id: issueRecoveryActions.id, evidence: issueRecoveryActions.evidence }) @@ -1104,9 +1104,10 @@ export function createPostgresWakeQueueAdapter(db: Db, deps: WakeQueuePostgresAd // next explicit wake adopts those messages atomically when it // queues a run. executionCancellationAcknowledged: - run.status === "cancelled" && + isAcknowledgedNativeReassignmentStop(run) || + (run.status === "cancelled" && (parseObject(run.resultJson?.executionCancellation).state === "acknowledged" || isAcknowledgedNativeStop(run)) && - !interruptedQueue, + !interruptedQueue), }; const preDrain = decidePreDrain(preDrainFacts); diff --git a/server/src/services/acknowledged-native-stop.test.ts b/server/src/services/acknowledged-native-stop.test.ts index a46a78a6342..f8772e5761e 100644 --- a/server/src/services/acknowledged-native-stop.test.ts +++ b/server/src/services/acknowledged-native-stop.test.ts @@ -1,5 +1,5 @@ import { expect, it } from "vitest"; -import { isAcknowledgedNativeStop } from "./acknowledged-native-stop.js"; +import { isAcknowledgedNativeReassignmentStop, isAcknowledgedNativeStop } from "./acknowledged-native-stop.js"; const run = { id: "run", companyId: "company", nativeIssueId: "issue", status: "cancelled", resultJson: { cancelledByActorType: "user", cancelledByUserId: "board", nativeCancellation: { schema: "paperclip.native-cancellation.v1", runId: "run", companyId: "company", issueId: "issue", @@ -15,3 +15,16 @@ it.each([{ runId: "other" }, { companyId: "other" }, { issueId: "other" }, { dis expect(isAcknowledgedNativeStop({ ...run, resultJson: { ...run.resultJson, nativeCancellation: { ...run.resultJson.nativeCancellation, ...change } } })).toBe(false); }); + +it("recognizes an audited handoff without granting operator Stop semantics", () => { + const handoff = { ...run, resultJson: { reassignmentStopRequested: true, nativeCancellation: run.resultJson.nativeCancellation } }; + expect(isAcknowledgedNativeReassignmentStop(handoff)).toBe(true); + expect(isAcknowledgedNativeStop(handoff)).toBe(false); + expect(isAcknowledgedNativeReassignmentStop({ ...handoff, status: "running" })).toBe(false); + expect(isAcknowledgedNativeReassignmentStop(run)).toBe(false); +}); +it.each([{ runId: "other" }, { companyId: "other" }, { issueId: "other" }, { dispatchState: "pending" }, + { scope: "subtree" }, { acknowledgementAuditId: undefined }])("requires an audited handoff receipt %j", change => { + expect(isAcknowledgedNativeReassignmentStop({ ...run, resultJson: { reassignmentStopRequested: true, + nativeCancellation: { ...run.resultJson.nativeCancellation, ...change } } })).toBe(false); +}); diff --git a/server/src/services/acknowledged-native-stop.ts b/server/src/services/acknowledged-native-stop.ts index 37efd1c84eb..bb6b7fa74c3 100644 --- a/server/src/services/acknowledged-native-stop.ts +++ b/server/src/services/acknowledged-native-stop.ts @@ -4,21 +4,34 @@ import { hasNativeLocalProcessStop } from "./native-local-process-stop.js"; import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; /** A server-recorded run-only Stop must not manufacture a recovery incident. */ -export function hasAcknowledgedNativeStopIntent(run: { +function hasAcknowledgedNativeRunStopReceipt(run: { id: string; companyId: string; status: string; nativeIssueId: string | null; resultJson: Record | null; }): boolean { const result = run.resultJson; const intent = result?.nativeCancellation as Record | undefined; - return result?.cancelledByActorType === "user" && - typeof result.cancelledByUserId === "string" && Boolean(result.cancelledByUserId) && - intent?.schema === "paperclip.native-cancellation.v1" && intent.runId === run.id && + return intent?.schema === "paperclip.native-cancellation.v1" && intent.runId === run.id && intent.companyId === run.companyId && intent.issueId === run.nativeIssueId && intent.scope === "run" && intent.reasonCode === "cancellation_run_only" && intent.dispatchState === "acknowledged" && intent.dispatched === true && typeof intent.intentAuditId === "string" && typeof intent.acknowledgementAuditId === "string"; } +export function hasAcknowledgedNativeStopIntent(run: Parameters[0]): boolean { + return run.resultJson?.cancelledByActorType === "user" && + typeof run.resultJson.cancelledByUserId === "string" && Boolean(run.resultJson.cancelledByUserId) && + hasAcknowledgedNativeRunStopReceipt(run); +} + +/** An intentional handoff retains task state and deferred work for its successor. */ +export function hasAcknowledgedNativeReassignmentStopIntent(run: Parameters[0]): boolean { + return run.resultJson?.reassignmentStopRequested === true && hasAcknowledgedNativeRunStopReceipt(run); +} + +export function isAcknowledgedNativeReassignmentStop(run: Parameters[0]): boolean { + return run.status === "cancelled" && hasAcknowledgedNativeReassignmentStopIntent(run); +} + export function isAcknowledgedNativeStop(run: Parameters[0]): boolean { return run.status === "cancelled" && hasAcknowledgedNativeStopIntent(run); } diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 57db2850dcb..40c0150dff1 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,7 +1,7 @@ import { readQueuedInteractionResponse } from "./queued-interaction-response.js"; import { AGENT_CHAT_DIRECTIVE, conversationReplay, isConversation, isConversationExecutionWake, isWaitingConversation, prepareConversationTurn, settleConversationTurn } from "./agent-conversations.js"; import { PROCESS_IDENTITY_RECORDED, recordNativeLocalProcessStop } from "./native-local-process-stop.js"; -import { hasAcknowledgedNativeStopIntent, isAcknowledgedNativeStop, acknowledgedNativeStopExecutionHasStopped } from "./acknowledged-native-stop.js"; +import { hasAcknowledgedNativeReassignmentStopIntent, hasAcknowledgedNativeStopIntent, isAcknowledgedNativeStop, acknowledgedNativeStopExecutionHasStopped } from "./acknowledged-native-stop.js"; import { legacyControllerBootId, legacyControllerClaim, renewLegacyControllerLease, hasLiveLegacyController, revokeExpiredLegacyController, watchLegacyControllerLease } from "./legacy-controller-lease.js"; import { completeTerminatedRemoteNativeSessionCleanup } from "../vendor/paperclip-runner/index.js"; import { hasRemoteTerminationReceipt, remoteExecutionHasStopped, remoteTerminationReceipt, stoppedRemoteCleanupScopes } from "./remote-execution-termination.js"; @@ -12849,7 +12849,7 @@ export function heartbeatService( // Teardown can beat the cancellation finalizer. Preserve the acknowledged // user intent instead of reporting an infrastructure interruption. - if (hasAcknowledgedNativeStopIntent(run)) terminalStatus = "cancelled"; + if (hasAcknowledgedNativeStopIntent(run) || hasAcknowledgedNativeReassignmentStopIntent(run)) terminalStatus = "cancelled"; const message = `run terminalized on environment lease release: heartbeat_runs.status was still ${run.status} at teardown`; // Match both "running" and "queued". A queued run has released its lease but @@ -28615,7 +28615,11 @@ export function heartbeatService( // additional durable fence before its existing cancellation path. if (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' 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 bbf29891101..90b849c2ee4 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -6059,11 +6059,13 @@ describe("native warm session supervision", () => { }); describe("native session bounded recovery", () => { - it("does not turn an acknowledged Stop before completion into a failure or a retry", async () => { + it.each(["operator", "reassignment"])("does not turn an acknowledged %s Stop before completion into a failure or a retry", async (source) => { const updates: Array<{ table: unknown; values: Record }> = []; const stop: Record = {}; state.execute.mockReset().mockImplementationOnce(async () => { - Object.assign(stop, { cancelledByActorType: "user", cancelledByUserId: "board", nativeCancellation: { + Object.assign(stop, { ...(source === "operator" + ? { cancelledByActorType: "user", cancelledByUserId: "board" } + : { reassignmentStopRequested: true }), nativeCancellation: { schema: "paperclip.native-cancellation.v1", ...execution.binding, scope: "run", reasonCode: "cancellation_run_only", dispatched: true, dispatchState: "acknowledged", intentAuditId: "intent", acknowledgementAuditId: "ack", } }); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 9e5347b566e..a0b3c1fab36 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -2,7 +2,7 @@ import { createLocalNativeQuestionBridge } from "./local-native-question-bridge. import { readVerifiedRemoteWorkspaceFile } from "./remote-deliverable-file.js"; import { copyBackCodexAuth } from "@paperclipai/adapter-codex-local/server"; import { nativeCompletionFeedback } from "./native-completion-feedback.js"; -import { hasAcknowledgedNativeStopIntent } from "../acknowledged-native-stop.js"; +import { hasAcknowledgedNativeReassignmentStopIntent, hasAcknowledgedNativeStopIntent } from "../acknowledged-native-stop.js"; import { stoppedCodexTurnIsTextOnly } from "./stopped-codex-turn.js"; import { prepareVerifiedRemoteProviderPack } from "./remote-provider-pack.js"; import { readNativeLocalProcessStop, PROCESS_START_REQUESTED } from "../native-local-process-stop.js"; @@ -8194,7 +8194,7 @@ async function executePaperclipNativeSessionWithinScope( eq(heartbeatRuns.agentId, input.execution.binding.agentId), eq(heartbeatRuns.nativeIssueId, input.execution.binding.issueId), )).limit(1); - if (stoppedRun && hasAcknowledgedNativeStopIntent(stoppedRun)) { + if (stoppedRun && (hasAcknowledgedNativeStopIntent(stoppedRun) || hasAcknowledgedNativeReassignmentStopIntent(stoppedRun))) { const [settled] = await input.db.update(nativeRunFinalizations).set({ phase: "terminal_failure", failureCode: "native_retry_cancelled", nextAttemptAt: null, leaseOwner: null, leaseExpiresAt: null, controlDeadlineAt: null, recoveryState: null,