From 80e34e3a40ea9b17db75d7688f32e5aa1f9f2606 Mon Sep 17 00:00:00 2001 From: Dotta Date: Mon, 21 Sep 2026 09:06:40 -0500 Subject: [PATCH] fix: cancel native startup before acknowledging operator stop Co-Authored-By: Paperclip --- doc/execution-semantics.md | 9 + .../native-session-executor.test.ts | 95 ++++ .../native-runtime/native-session-executor.ts | 428 ++++++++++-------- 3 files changed, 332 insertions(+), 200 deletions(-) diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 5c2cae81a3e..a454fdefc76 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1093,6 +1093,15 @@ controller lease in the same transaction that creates the native coordinator. The native executor rechecks cancellation and terminal status when claiming the coordinator, before starting or attaching a provider. +Run-only Stop also covers the interval after the coordinator claim and before +the provider session publishes its handle. Stop retains its pending audited +intent and waits up to 30 seconds for that startup to settle. A published +session receives cancellation before prompt submission; only real dispatch +sets `dispatched: true`. A deadline leaves the intent pending and the late +session remains fenced and is closed. Stop acknowledgement alone does not +certify cleanup: the existing process and environment receipts still govern +admission of the next message. + A cancelled startup can continue from a newer authenticated user message after cleanup. The server requires either its explicit before-selection fence or an unclaimed native coordinator (zero attempts and controller generations, no 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 e244964226d..956d78d0591 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -4903,6 +4903,101 @@ describe("native session cancellation", () => { }); }); + it("waits for an in-flight startup handle before acknowledging Stop", async () => { + const root = await mkdtemp(join(tmpdir(), "native-startup-stop-")); + const previous = process.env.PAPERCLIP_RUNNER_STATE_DIR; + process.env.PAPERCLIP_RUNNER_STATE_DIR = root; + let open!: () => void, started!: () => void, finish!: () => void; + const opening = new Promise(resolve => { open = resolve; }); + const admitted = new Promise(resolve => { started = resolve; }); + const finished = new Promise(resolve => { finish = resolve; }); + const submitTurn = vi.fn(); + state.execute.mockReset().mockImplementationOnce(async (options) => { + started(); + await opening; + await options.onSession({ cancel: state.cancel }); + submitTurn(); + await finished; + await options.onSession(null); + return { + result: { summary: "cancelled" }, terminal: { runTerminalState: "cancelled" }, + turnId: "turn", normalizedSessionId: "session", providerSessionId: null, + driverKind: "test", driverVersion: "1", nativeEventCount: 0, + highestContiguousSourceSeq: 0, + }; + }); + const running = executePaperclipNativeSession({ + db: leaseDb(), execution, runnerInstanceId: "runner", useRunnerd: true, + }); + const outcome = running.catch(error => error); + const persistence = cancellationDb(); + let stopping: ReturnType | undefined; + let acknowledged = false; + try { + await admitted; + stopping = cancelNativeSession(execution.binding.runId, "operator Stop during startup", { + db: persistence.db, scope: "run", + }); + void stopping.then(() => { acknowledged = true; }); + await new Promise(resolve => setImmediate(resolve)); + expect(acknowledged).toBe(false); + expect(persistence.getResultJson().nativeCancellation).toMatchObject({ dispatchState: "pending" }); + open(); + await expect(stopping).resolves.toMatchObject({ dispatched: true }); + expect(state.cancel).toHaveBeenCalledOnce(); + expect(persistence.getResultJson().nativeCancellation).toMatchObject({ + dispatchState: "acknowledged", dispatched: true, + }); + finish(); + await outcome; + expect(submitTurn).not.toHaveBeenCalled(); + } finally { + open(); finish(); + await outcome; + await stopping; + 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("fences a late startup even when Stop reaches its acknowledgement deadline", async () => { + const root = await mkdtemp(join(tmpdir(), "native-late-startup-stop-")); + 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.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 outcome = executePaperclipNativeSession({ db: leaseDb(), execution, runnerInstanceId: "runner", useRunnerd: true }).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); + // Let the durable intent commit before advancing the startup deadline. + await new Promise(resolve => setImmediate(resolve)); + await vi.advanceTimersByTimeAsync(30_000); + expect(await stopping).toMatchObject({ message: "native_cancellation_startup_not_ready" }); + expect(persistence.getResultJson().nativeCancellation).toMatchObject({ dispatchState: "pending" }); + vi.useRealTimers(); + open(); await outcome; + expect(state.cancel).toHaveBeenCalledOnce(); + expect(submitTurn).not.toHaveBeenCalled(); + } 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 5abfbe1b55f..6be215a5fda 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -209,6 +209,8 @@ const nativeRunsDetachingForRestart = new Set(); type NativeSessionStartup = { promise: Promise; resolve: (session: ActiveNativeSession | null) => void; + stopRequested?: boolean; + cancellationSettled?: Promise; }; const nativeSessionStartups = new Map(); @@ -217,7 +219,7 @@ function detachActiveNativeSessionForRestart(active: ActiveNativeSession) { return active.restartDetach; } -async function waitForNativeStartupForRestart(runId: string) { +async function waitForNativeSessionStartup(runId: string, failureCode = "native_restart_startup_not_ready") { const startup = nativeSessionStartups.get(runId); if (!startup) return null; let timer: ReturnType | undefined; @@ -225,7 +227,7 @@ async function waitForNativeStartupForRestart(runId: string) { return await Promise.race([ startup.promise, new Promise((_resolve, reject) => { - timer = setTimeout(() => reject(new Error("native_restart_startup_not_ready")), 30_000); + timer = setTimeout(() => reject(new Error(failureCode)), 30_000); }), ]); } finally { @@ -245,7 +247,7 @@ export async function detachNativeSessionsForRestart( const unsupportedRunIds: string[] = []; for (const runId of new Set(runIds)) { nativeRunsDetachingForRestart.add(runId); - const active = activeNativeSessions.get(runId) ?? await waitForNativeStartupForRestart(runId); + const active = activeNativeSessions.get(runId) ?? await waitForNativeSessionStartup(runId); if (!active) { inactiveRunIds.push(runId); continue; @@ -6289,6 +6291,7 @@ export async function cancelNativeSession( auditId: string | null; } > { + const runStop = (options?.scope ?? "run") === "run"; let decision: NativeStatusDecision | null = null; let decisionContext: { companyId: string; @@ -6554,7 +6557,8 @@ export async function cancelNativeSession( decisionId = intent.decisionId; recoveringCancellationIntent = intent.existing; priorCoordinatorDecisionIdAtIntent = intent.priorCoordinatorDecisionId; - if (intent.acknowledged) { + if (intent.acknowledged && (!runStop || intent.dispatched || + (!nativeSessionStartups.has(runId) && !activeNativeSessions.has(runId)))) { return { dispatched: intent.dispatched, decision, @@ -6563,150 +6567,208 @@ export async function cancelNativeSession( }; } } - const active = activeNativeSessions.get(runId); - let dispatched = false; - if (active) { - dispatched = true; - if (!active.cancelRequested) { - active.cancelRequested = true; - try { - if (active.session.cancel) { - const cancellationAbort = new AbortController(); - const cleanup = active.session.cancel({ - reason, - signal: cancellationAbort.signal, - }).cleanup; - let cleanupTimer: ReturnType | undefined; - const settled = await Promise.race([ - cleanup.then( - () => true, - () => true, - ), - new Promise((resolve) => { - cleanupTimer = setTimeout( - () => resolve(false), - NATIVE_SESSION_CANCELLATION_CLEANUP_GRACE_MS, - ); - }), - ]); - if (cleanupTimer) clearTimeout(cleanupTimer); - if (!settled) { - cancellationAbort.abort( - new Error("native session cancellation cleanup timed out"), - ); - void cleanup.catch(() => undefined); - } - } else if (active.session.interrupt) - await active.session.interrupt({ reason }); - } catch (error) { - active.cancelRequested = false; - throw error; - } - } + // Startup already owns a provider lifetime even before publishing its handle. + // Do not acknowledge a no-op Stop and let that owner submit a turn afterwards. + const startup = runStop && !activeNativeSessions.has(runId) ? nativeSessionStartups.get(runId) : undefined; + let settleStartupCancellation: (() => void) | undefined; + if (startup) { + startup.stopRequested = true; + startup.cancellationSettled = new Promise(resolve => { settleStartupCancellation = resolve; }); } - if (!options) return dispatched; - - if (decision && decisionContext) { - const cancellationDecision = decision; - const cancellationContext = decisionContext; - if (!cancellationIntentId || !auditId) - throw new Error("native_cancellation_intent_audit_missing"); - if ( - recoveringCancellationIntent && - cancellationContext.coordinatorDecisionId !== - priorCoordinatorDecisionIdAtIntent - ) { - decisionId ??= cancellationContext.coordinatorDecisionId; - } - if ( - cancellationContext.assessmentId && - cancellationDecision.reasonCode !== null && - !decisionId - ) { - const committed = await commitNativeStatusDecision({ - db: options.db, - companyId: cancellationContext.companyId, - issueId: cancellationContext.issueId, - runId, - assessmentId: cancellationContext.assessmentId, - priorStatus: cancellationContext.priorStatus, - priorStatusVersion: cancellationContext.priorStatusVersion, - priorDecisionId: cancellationContext.priorDecisionId, - decision: cancellationDecision, - }); - decisionId = committed.decision.id; - } - const acknowledgement = await options.db.transaction(async (tx) => { - const lockedRun = await tx - .select({ - agentId: heartbeatRuns.agentId, - companyId: heartbeatRuns.companyId, - nativeIssueId: heartbeatRuns.nativeIssueId, - resultJson: heartbeatRuns.resultJson, - runtimeMode: heartbeatRuns.runtimeMode, - }) - .from(heartbeatRuns) - .where(eq(heartbeatRuns.id, runId)) - .for("update") - .limit(1) - .then((rows) => rows[0] ?? null); - if ( - !lockedRun || - lockedRun.runtimeMode !== "native" || - lockedRun.companyId !== cancellationContext.companyId || - lockedRun.agentId !== cancellationContext.agentId || - lockedRun.nativeIssueId !== cancellationContext.issueId - ) { - throw new Error("native_cancellation_binding_changed"); - } - const coordinator = await tx - .select({ runId: nativeRunFinalizations.runId }) - .from(nativeRunFinalizations) - .where( - and( - eq(nativeRunFinalizations.runId, runId), - eq(nativeRunFinalizations.companyId, cancellationContext.companyId), - eq(nativeRunFinalizations.issueId, cancellationContext.issueId), - ), - ) - .limit(1) - .then((rows) => rows[0] ?? null); - if (!coordinator) - throw new Error("native_cancellation_coordinator_missing"); - - const resultJson = record(lockedRun.resultJson); - const intent = record(resultJson.nativeCancellation); - const matchingIntent = - intent.schema === "paperclip.native-cancellation.v1" && - intent.intentId === cancellationIntentId && - intent.intentAuditId === auditId && - intent.companyId === cancellationContext.companyId && - intent.runId === runId && - intent.issueId === cancellationContext.issueId; - if (!matchingIntent) - throw new Error("native_cancellation_intent_conflict"); - if (intent.dispatchState === "acknowledged") { - return { - publication: null, - decisionId: - typeof intent.decisionId === "string" - ? intent.decisionId - : decisionId, - }; + try { + const active = activeNativeSessions.get(runId) ?? + (startup ? await waitForNativeSessionStartup(runId, "native_cancellation_startup_not_ready") : null); + let dispatched = false; + if (active) { + dispatched = true; + if (!active.cancelRequested) { + active.cancelRequested = true; + try { + if (active.session.cancel) { + const cancellationAbort = new AbortController(); + const cleanup = active.session.cancel({ + reason, + signal: cancellationAbort.signal, + }).cleanup; + let cleanupTimer: ReturnType | undefined; + const settled = await Promise.race([ + cleanup.then( + () => true, + () => true, + ), + new Promise((resolve) => { + cleanupTimer = setTimeout( + () => resolve(false), + NATIVE_SESSION_CANCELLATION_CLEANUP_GRACE_MS, + ); + }), + ]); + if (cleanupTimer) clearTimeout(cleanupTimer); + if (!settled) { + cancellationAbort.abort( + new Error("native session cancellation cleanup timed out"), + ); + void cleanup.catch(() => undefined); + } + } else if (active.session.interrupt) + await active.session.interrupt({ reason }); + } catch (error) { + active.cancelRequested = false; + throw error; + } } + } + if (!options) return dispatched; + if (decision && decisionContext) { + const cancellationDecision = decision; + const cancellationContext = decisionContext; + if (!cancellationIntentId || !auditId) + throw new Error("native_cancellation_intent_audit_missing"); if ( - options.replacementAccepted && - cancellationDecision.effects.some( - (effect) => effect.kind === "accept_replacement_turn", - ) + recoveringCancellationIntent && + cancellationContext.coordinatorDecisionId !== + priorCoordinatorDecisionIdAtIntent ) { - await tx + decisionId ??= cancellationContext.coordinatorDecisionId; + } + if ( + cancellationContext.assessmentId && + cancellationDecision.reasonCode !== null && + !decisionId + ) { + const committed = await commitNativeStatusDecision({ + db: options.db, + companyId: cancellationContext.companyId, + issueId: cancellationContext.issueId, + runId, + assessmentId: cancellationContext.assessmentId, + priorStatus: cancellationContext.priorStatus, + priorStatusVersion: cancellationContext.priorStatusVersion, + priorDecisionId: cancellationContext.priorDecisionId, + decision: cancellationDecision, + }); + decisionId = committed.decision.id; + } + const acknowledgement = await options.db.transaction(async (tx) => { + const lockedRun = await tx + .select({ + agentId: heartbeatRuns.agentId, + companyId: heartbeatRuns.companyId, + nativeIssueId: heartbeatRuns.nativeIssueId, + resultJson: heartbeatRuns.resultJson, + runtimeMode: heartbeatRuns.runtimeMode, + }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .for("update") + .limit(1) + .then((rows) => rows[0] ?? null); + if ( + !lockedRun || + lockedRun.runtimeMode !== "native" || + lockedRun.companyId !== cancellationContext.companyId || + lockedRun.agentId !== cancellationContext.agentId || + lockedRun.nativeIssueId !== cancellationContext.issueId + ) { + throw new Error("native_cancellation_binding_changed"); + } + const coordinator = await tx + .select({ runId: nativeRunFinalizations.runId }) + .from(nativeRunFinalizations) + .where( + and( + eq(nativeRunFinalizations.runId, runId), + eq(nativeRunFinalizations.companyId, cancellationContext.companyId), + eq(nativeRunFinalizations.issueId, cancellationContext.issueId), + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null); + if (!coordinator) + throw new Error("native_cancellation_coordinator_missing"); + + const resultJson = record(lockedRun.resultJson); + const intent = record(resultJson.nativeCancellation); + const matchingIntent = + intent.schema === "paperclip.native-cancellation.v1" && + intent.intentId === cancellationIntentId && + intent.intentAuditId === auditId && + intent.companyId === cancellationContext.companyId && + intent.runId === runId && + intent.issueId === cancellationContext.issueId; + if (!matchingIntent) + throw new Error("native_cancellation_intent_conflict"); + if (intent.dispatchState === "acknowledged" && (intent.dispatched === true || !dispatched)) { + return { + publication: null, + decisionId: + typeof intent.decisionId === "string" + ? intent.decisionId + : decisionId, + }; + } + + if ( + options.replacementAccepted && + cancellationDecision.effects.some( + (effect) => effect.kind === "accept_replacement_turn", + ) + ) { + await tx + .update(heartbeatRuns) + .set({ + status: "running", + continuationAttempt: sql`${heartbeatRuns.continuationAttempt} + 1`, + nextAction: "Accept a replacement native turn on the existing run.", + updatedAt: new Date(), + }) + .where( + and( + eq(heartbeatRuns.id, runId), + eq(heartbeatRuns.companyId, cancellationContext.companyId), + eq(heartbeatRuns.agentId, cancellationContext.agentId), + eq(heartbeatRuns.nativeIssueId, cancellationContext.issueId), + ), + ); + } + const activity = await persistActivity(tx as unknown as Db, { + companyId: cancellationContext.companyId, + actorType: "system", + actorId: "native-session-cancellation", + action: "native.cancellation_dispatch_acknowledged", + entityType: "heartbeat_run", + entityId: runId, + agentId: cancellationContext.agentId, + runId, + issueId: cancellationContext.issueId, + details: { + intentId: cancellationIntentId, + intentAuditId: auditId, + scope: options.scope ?? "run", + reasonCode: cancellationDecision.reasonCode, + effects: cancellationDecision.effects.map((effect) => effect.kind), + dispatched, + decisionId, + }, + }); + const acknowledgementAuditId = activity.activity?.id ?? null; + if (!acknowledgementAuditId) + throw new Error("native_cancellation_ack_audit_missing"); + const cancellationWrite = await tx .update(heartbeatRuns) .set({ - status: "running", - continuationAttempt: sql`${heartbeatRuns.continuationAttempt} + 1`, - nextAction: "Accept a replacement native turn on the existing run.", + resultJson: { + ...resultJson, + nativeCancellation: { + ...intent, + dispatchState: "acknowledged", + dispatched, + decisionId, + acknowledgementAuditId, + acknowledgedAt: new Date().toISOString(), + }, + }, updatedAt: new Date(), }) .where( @@ -6716,66 +6778,21 @@ export async function cancelNativeSession( eq(heartbeatRuns.agentId, cancellationContext.agentId), eq(heartbeatRuns.nativeIssueId, cancellationContext.issueId), ), - ); - } - const activity = await persistActivity(tx as unknown as Db, { - companyId: cancellationContext.companyId, - actorType: "system", - actorId: "native-session-cancellation", - action: "native.cancellation_dispatch_acknowledged", - entityType: "heartbeat_run", - entityId: runId, - agentId: cancellationContext.agentId, - runId, - issueId: cancellationContext.issueId, - details: { - intentId: cancellationIntentId, - intentAuditId: auditId, - scope: options.scope ?? "run", - reasonCode: cancellationDecision.reasonCode, - effects: cancellationDecision.effects.map((effect) => effect.kind), - dispatched, - decisionId, - }, + ) + .returning({ id: heartbeatRuns.id }) + .then((rows) => rows[0] ?? null); + if (!cancellationWrite) + throw new Error("native_cancellation_binding_changed"); + return { publication: activity.publication, decisionId }; }); - const acknowledgementAuditId = activity.activity?.id ?? null; - if (!acknowledgementAuditId) - throw new Error("native_cancellation_ack_audit_missing"); - const cancellationWrite = await tx - .update(heartbeatRuns) - .set({ - resultJson: { - ...resultJson, - nativeCancellation: { - ...intent, - dispatchState: "acknowledged", - dispatched, - decisionId, - acknowledgementAuditId, - acknowledgedAt: new Date().toISOString(), - }, - }, - updatedAt: new Date(), - }) - .where( - and( - eq(heartbeatRuns.id, runId), - eq(heartbeatRuns.companyId, cancellationContext.companyId), - eq(heartbeatRuns.agentId, cancellationContext.agentId), - eq(heartbeatRuns.nativeIssueId, cancellationContext.issueId), - ), - ) - .returning({ id: heartbeatRuns.id }) - .then((rows) => rows[0] ?? null); - if (!cancellationWrite) - throw new Error("native_cancellation_binding_changed"); - return { publication: activity.publication, decisionId }; - }); - decisionId = acknowledgement.decisionId; - if (acknowledgement.publication) - publishActivity(acknowledgement.publication); + decisionId = acknowledgement.decisionId; + if (acknowledgement.publication) + publishActivity(acknowledgement.publication); + } + return { dispatched, decision, decisionId, auditId }; + } finally { + settleStartupCancellation?.(); } - return { dispatched, decision, decisionId, auditId }; } /** Authenticated cancellation scope projected through the shared arbiter. */ @@ -8164,7 +8181,18 @@ async function executePaperclipNativeSessionWithinScope( if (nativeRunsDetachingForRestart.has(input.execution.binding.runId)) { if (session.detachControllerForRestart) await detachActiveNativeSessionForRestart(active); } - nativeSessionStartups.get(input.execution.binding.runId)?.resolve(active); + const startup = nativeSessionStartups.get(input.execution.binding.runId); + startup?.resolve(active); + if (startup?.stopRequested) { + // Publication wakes the Stop caller; wait for its durable ACK + // before unwinding. No provider turn may be submitted between + // session startup and the execution-owned cleanup below. + await startup.cancellationSettled; + // If startup exceeded the caller's deadline, its late handle + // must still be cancelled instead of escaping the Stop fence. + await cancelNativeSession(input.execution.binding.runId, "Stop requested during native startup"); + throw new Error("native_finalization_missing: session returned no semantic result"); + } } else { liveQuestions.close(); activeNativeSessions.delete(input.execution.binding.runId);