fix: cancel native startup before acknowledging operator stop

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-09-21 09:06:40 -05:00
1 parent 3d6140ca5d
commit 80e34e3a40
3 files changed
+332 -200

No files matched your search

+9
View File
@@ -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
@@ -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<void>(resolve => { open = resolve; });
const admitted = new Promise<void>(resolve => { started = resolve; });
const finished = new Promise<void>(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<typeof cancelNativeSession> | 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<void>(resolve => { open = resolve; });
const admitted = new Promise<void>(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(),
@@ -209,6 +209,8 @@ const nativeRunsDetachingForRestart = new Set<string>();
type NativeSessionStartup = {
promise: Promise<ActiveNativeSession | null>;
resolve: (session: ActiveNativeSession | null) => void;
stopRequested?: boolean;
cancellationSettled?: Promise<void>;
};
const nativeSessionStartups = new Map<string, NativeSessionStartup>();
@@ -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<typeof setTimeout> | undefined;
@@ -225,7 +227,7 @@ async function waitForNativeStartupForRestart(runId: string) {
return await Promise.race([
startup.promise,
new Promise<never>((_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<typeof setTimeout> | undefined;
const settled = await Promise.race([
cleanup.then(
() => true,
() => true,
),
new Promise<false>((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<void>(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<typeof setTimeout> | undefined;
const settled = await Promise.race([
cleanup.then(
() => true,
() => true,
),
new Promise<false>((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);