fix: fence native startup against cancellation (#13316)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Users can pause a task while its runner prepares to start. > - Cancellation must prevent preparation from creating new execution authority. > - Native runtime selection could run after cancellation and leave an unclaimed recovery coordinator. > - Saved user messages then waited for recovery that had no eligible worker. > - This pull request fences startup and lets explicit user continuation settle verified, unclaimed startup state. > - Tasks can continue after cleanup while keeping provider ownership and execution safeguards. ## Linked Issues or Issue Description Refs #13285 for startup controller leases and #13270 for explicit continuation and saved-message recovery. Related #13293 covers retained processes that actually started; this change covers cancellation before the native provider claim. Related #13315 covers legacy queued-message delivery. **What happened?** Pausing a task during startup could cancel its heartbeat before native runtime selection. Stale preparation then created an observed native coordinator on the cancelled run. The coordinator had no provider result or eligible recovery worker. A later Continue message stayed queued indefinitely. **Expected behavior** Cancellation fences native startup. After verified cleanup, a newer user message starts one fresh conversation turn. An unverified execution keeps its hold and a clear explanation. **Steps to reproduce** 1. Start a task with the native runner and delay startup preparation before runtime selection. 2. Pause the task, then release preparation. 3. Resume the task and send Continue. 4. Before this fix, native selection can persist after cancellation and block the saved message. 5. Repeat from persisted cancelled startup state after a server restart. **Paperclip version or commit** Reproduced against master at `586b5ec82` with isolated PostgreSQL regression fixtures. **Deployment mode** Self-hosted server built from source, with Paperclip Runner. ## What Changed - Serialize the cancellation fence and native runtime selection on the run row. Revalidate the startup controller lease. - Refresh the runtime before dispatching cancellation, and reject terminal or cancelled runs at the native provider claim. - Recognize never-claimed coordinators only after startup and environment cleanup are verified. Reject process, provider, owner, and conflicting launch evidence. - Settle that coordinator atomically with a new authenticated user turn. Preserve history, unknown outcomes, and attempt counts. - Reuse the saved-message worker after restart and retain pause, budget, approval, and ownership gates. - Add startup, restart, duplicate-admission, and negative-proof regressions. Document the rule. ## Verification - All 687 tests pass across the complete heartbeat recovery, explicit continuation, and native session executor suites on the rebased branch. - Six focused race regressions also pass: cancellation before and after native selection, Stop racing adapter registration, process termination during a database failure, and a run finishing during cancellation. - `pnpm -r typecheck` and `pnpm build` pass after rebasing on master at `ab15aff39`. - Greptile is 5/5 on `f3dea2ab27facdf0360b56172ebd3e5219e25538`, with no open review threads. Policy and security checks pass. - All 32 CI checks pass on the final head, including the full server/workspace test matrix, all three browser shards, runner verification, typechecks, build, and canary dry run. The two optional Storybook jobs are skipped. [CI run](https://github.com/paperclipai/paperclip/actions/runs/34697945514). - Local full-suite limitation: the broad `pnpm test:run` attempt reported two failures outside the changed area after unusually long test durations (about 65 seconds for supporting skill-file saves and 933 seconds for setup-token login). Both cases passed isolated reruns, with no code changes. The broad local run was stopped after CI completed successfully; no clean full local-suite pass is claimed. ## Risks Cancellation and startup overlap. The run and coordinator locks provide the authority fence; cleanup and process evidence provide the containment proof. Historical runs without sufficient evidence remain blocked. A saved user message authorizes a fresh turn, not automatic replay. No schema migration or dependency change. ## Model Used OpenAI GPT-6 through Codex, using reasoning, repository inspection, code execution, and tests. This session does not expose the exact backend revision or context-window size. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Dotta
GitHub
parent
9132e8279f
commit
0e14c61da7
|
|
@ -987,3 +987,25 @@ For a board operator, the intended meaning is:
|
|||
- blockers explain waiting
|
||||
|
||||
That is the execution contract Paperclip should present to operators.
|
||||
|
||||
### Cancellation during native startup
|
||||
|
||||
Cancellation records a preparation fence while holding the run row lock. Native
|
||||
runtime selection checks that fence, the running status, and the current startup
|
||||
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.
|
||||
|
||||
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
|
||||
controller, lease, or result). It also checks for contradictory launch/process
|
||||
evidence and verifies local cleanup or exact remote termination receipts. The
|
||||
preparer must have finished or its startup lease must have expired. A missing
|
||||
PID alone does not establish this proof.
|
||||
|
||||
The existing bounded saved-message worker rechecks this proof after restart.
|
||||
Admission atomically settles an unclaimed coordinator and admits one fresh turn,
|
||||
preserving history, unknown action outcomes, and attempt counts. Pauses, approvals,
|
||||
budgets, task ownership, and terminal task status still gate admission. No
|
||||
automatic provider replay is authorized by a cancelled startup.
|
||||
|
|
|
|||
|
|
@ -2537,6 +2537,69 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
|||
}
|
||||
}
|
||||
|
||||
it("fences native selection when cancellation wins during preparation", async () => {
|
||||
await withTempPaperclipHome(async () => {
|
||||
const { agentId, issueId, runId } = await seedQueuedIssueRunFixture();
|
||||
await db.update(agents).set({ adapterType: "paperclip_runner",
|
||||
adapterConfig: { provider: "codex", model: "gpt-5.6-luna" },
|
||||
}).where(eq(agents.id, agentId));
|
||||
const factory = vi.fn(() => { throw new Error("provider must not start"); });
|
||||
let reachedSelection = false;
|
||||
const heartbeat = heartbeatService(db, {
|
||||
nativeSessionBackendFactory: factory,
|
||||
beforeNativeRuntimeSelection: async id => {
|
||||
reachedSelection = true;
|
||||
await heartbeat.cancelRun(id);
|
||||
},
|
||||
});
|
||||
await heartbeat.resumeQueuedRuns();
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
expect(reachedSelection).toBe(true);
|
||||
expect(await heartbeat.getRun(runId)).toMatchObject({ status: "cancelled", runtimeMode: "legacy",
|
||||
runtimeModeResolvedAt: null, nativeSessionId: null,
|
||||
resultJson: { startupCancellation: { beforeNativeSelection: true },
|
||||
startupPreparationSettledAt: expect.any(String) },
|
||||
});
|
||||
expect(await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId))).toHaveLength(0);
|
||||
expect(factory).not.toHaveBeenCalled();
|
||||
expect(mockAdapterExecute).not.toHaveBeenCalled();
|
||||
const [task] = await db.select().from(issues).where(eq(issues.id, issueId));
|
||||
expect(task.executionRunId).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
it("does not dispatch when cancellation wins after native selection", async () => {
|
||||
await withTempPaperclipHome(async () => {
|
||||
const { agentId, issueId, runId } = await seedQueuedIssueRunFixture();
|
||||
await db.update(agents).set({ adapterType: "paperclip_runner",
|
||||
adapterConfig: { provider: "codex", model: "gpt-5.6-luna" },
|
||||
}).where(eq(agents.id, agentId));
|
||||
await db.update(heartbeatRuns).set({ invocationSource: "automation" }).where(eq(heartbeatRuns.id, runId));
|
||||
const factory = vi.fn(() => { throw new Error("provider must not start"); });
|
||||
let reachedDispatch = false;
|
||||
const heartbeat = heartbeatService(db, {
|
||||
nativeSessionBackendFactory: factory,
|
||||
beforeChatControlRecoveryCheck: async ({ stage, runId: id }) => {
|
||||
if (stage !== "dispatch") return;
|
||||
reachedDispatch = true;
|
||||
expect((await heartbeat.getRun(id))?.runtimeMode).toBe("native");
|
||||
await heartbeat.cancelRun(id);
|
||||
},
|
||||
});
|
||||
await heartbeat.resumeQueuedRuns();
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
expect(reachedDispatch).toBe(true);
|
||||
expect(await heartbeat.getRun(runId)).toMatchObject({ status: "cancelled", runtimeMode: "native",
|
||||
resultJson: { startupPreparationSettledAt: expect.any(String) },
|
||||
});
|
||||
expect(factory).not.toHaveBeenCalled();
|
||||
const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId));
|
||||
expect(coordinator).toMatchObject({ attempt: 0, leaseOwner: null });
|
||||
const [task] = await db.select().from(issues).where(eq(issues.id, issueId));
|
||||
expect(task.executionRunId).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
it("dispatches local native external chat inside the server-selected task root", async () => {
|
||||
await withTempPaperclipHome(async () => {
|
||||
const { companyId, agentId, issueId, runId } =
|
||||
|
|
|
|||
|
|
@ -0,0 +1,46 @@
|
|||
import { and, eq, inArray, isNotNull, or } from "drizzle-orm";
|
||||
import { environmentLeases, heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db";
|
||||
import { claimedAdapterType } from "./conversation-continuation.js";
|
||||
import { PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED } from "./native-local-process-stop.js";
|
||||
import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js";
|
||||
|
||||
type Run = typeof heartbeatRuns.$inferSelect;
|
||||
type Coordinator = typeof nativeRunFinalizations.$inferSelect;
|
||||
|
||||
/** Caller holds the coordinator and run locks when using this proof to admit
|
||||
* work. Attempt zero is a durable never-claimed receipt: every native executor
|
||||
* commits its first claim before it can start or attach a provider. */
|
||||
export async function isCancelledNativeStartup(db: Db, run: Run, coordinator: Coordinator | undefined) {
|
||||
if (run.status !== "cancelled" || !run.finishedAt || run.processPid || run.processGroupId ||
|
||||
run.processStartedAt || run.sessionIdAfter) return false;
|
||||
const cancellation = run.resultJson?.startupCancellation as Record<string, unknown> | undefined;
|
||||
const beforeSelection = run.runtimeMode === "legacy" && !run.runtimeModeResolvedAt &&
|
||||
!run.nativeSessionId && !coordinator && claimedAdapterType(run) === "paperclip_runner" &&
|
||||
cancellation?.beforeNativeSelection === true;
|
||||
const neverClaimed = run.runtimeMode === "native" && coordinator &&
|
||||
["observed", "terminal_failure"].includes(coordinator.phase) && coordinator.attempt === 0 &&
|
||||
coordinator.controllerGeneration === 0 && !coordinator.controllerBootId &&
|
||||
!coordinator.controllerPid && !coordinator.leaseOwner && !coordinator.leaseExpiresAt &&
|
||||
!coordinator.resultId && !coordinator.failureDetail?.successorRunId;
|
||||
if (!beforeSelection && !neverClaimed) return false;
|
||||
const settled = typeof run.resultJson?.startupPreparationSettledAt === "string";
|
||||
// The old preparer can still be unwinding even though the run is terminal.
|
||||
if (!settled && run.controllerLeaseExpiresAt && run.controllerLeaseExpiresAt > new Date()) return false;
|
||||
const leases = await db.select().from(environmentLeases).where(and(
|
||||
eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id),
|
||||
));
|
||||
if ((!settled && leases.length === 0) || leases.some(lease =>
|
||||
lease.provider === "local"
|
||||
? !lease.releasedAt || lease.status === "pending_cleanup" || lease.cleanupStatus === "failed"
|
||||
: !hasRemoteTerminationReceipt(lease))) return false;
|
||||
// Reject contradictory retained evidence, including a crash after a launch
|
||||
// request but before the PID callback. Provider events never certify a stop.
|
||||
const [execution] = await db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and(
|
||||
eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id),
|
||||
or(isNotNull(heartbeatRunEvents.sourceEventId),
|
||||
inArray(heartbeatRunEvents.eventType, [PROCESS_START_REQUESTED, PROCESS_IDENTITY_RECORDED,
|
||||
"harness.ready", "session.started", "session.resumed", "session.updated", "turn.started",
|
||||
"provider.event", "provider.rpc_result", "tool.execution.started"])),
|
||||
)).limit(1);
|
||||
return !execution;
|
||||
}
|
||||
|
|
@ -49,6 +49,72 @@ const support = await getEmbeddedPostgresTestSupport();
|
|||
agentId: f.agentId, status: "queued", contextSnapshot: { issueId: f.issueId, previousRunId: result.previousRunId, forceFreshSession: true } });
|
||||
return result;
|
||||
});
|
||||
async function seedCancelledStartup() {
|
||||
const f = await seed();
|
||||
await db.update(heartbeatRuns).set({ status: "cancelled", processPid: null,
|
||||
startedAt: new Date("2026-09-11T09:59:59Z"),
|
||||
runtimeModeResolvedAt: new Date("2026-09-11T10:00:01Z"),
|
||||
controllerBootId: randomUUID(), controllerLeaseExpiresAt: new Date("2026-09-11T10:01:00Z"),
|
||||
}).where(eq(heartbeatRuns.id, f.sourceRunId));
|
||||
await db.update(nativeRunFinalizations).set({ phase: "observed", attempt: 0,
|
||||
failureDetail: null,
|
||||
}).where(eq(nativeRunFinalizations.runId, f.sourceRunId));
|
||||
await db.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: f.sourceRunId,
|
||||
provider: "local", status: "released", releasedAt: new Date("2026-09-11T10:00:02Z"),
|
||||
cleanupStatus: "succeeded", leasePolicy: "ephemeral" });
|
||||
return f;
|
||||
}
|
||||
|
||||
it("settles a cancelled unclaimed coordinator after restart and admits one user successor", async () => {
|
||||
const f = await seedCancelledStartup();
|
||||
expect(await admit(f, true)).toMatchObject({ previousRunId: f.sourceRunId });
|
||||
expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId)))[0].phase).toBe("observed");
|
||||
const results = await Promise.all([admit(f), admit(f)]);
|
||||
expect(results.filter(Boolean)).toHaveLength(1);
|
||||
const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId));
|
||||
expect(coordinator).toMatchObject({ phase: "terminal_failure", attempt: 0,
|
||||
failureCode: "native_startup_cancelled", failureDetail: { replacementDenied: "explicit_user_continuation" } });
|
||||
const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
|
||||
expect(action.evidence.automaticRecovery).toMatchObject({ actionOutcome: "unknown", replay: "explicit_user_continuation" });
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
|
||||
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId)))[0].status).toBe("cancelled");
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.successorRunId))).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("continues native-runner preparation cancelled before runtime selection", async () => {
|
||||
const f = await seedCancelledStartup();
|
||||
await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId));
|
||||
await db.update(heartbeatRuns).set({ runtimeMode: "legacy", runtimeModeResolvedAt: null, nativeIssueId: null,
|
||||
runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } },
|
||||
resultJson: { startupCancellation: { beforeNativeSelection: true }, startupPreparationSettledAt: new Date().toISOString() },
|
||||
}).where(eq(heartbeatRuns.id, f.sourceRunId));
|
||||
expect(await admit(f)).toMatchObject({ previousRunId: f.sourceRunId });
|
||||
});
|
||||
|
||||
it.each(["attempt", "generation", "controller", "lease", "process", "launch", "provider", "cleanup", "remote", "preparing", "closed", "reassigned"])(
|
||||
"retains cancellation safeguards with %s evidence", async kind => {
|
||||
const f = await seedCancelledStartup();
|
||||
if (kind === "attempt") await db.update(nativeRunFinalizations).set({ attempt: 1 }).where(eq(nativeRunFinalizations.runId, f.sourceRunId));
|
||||
if (kind === "generation") await db.update(nativeRunFinalizations).set({ controllerGeneration: 1 }).where(eq(nativeRunFinalizations.runId, f.sourceRunId));
|
||||
if (kind === "controller") await db.update(nativeRunFinalizations).set({ controllerBootId: "old-owner" }).where(eq(nativeRunFinalizations.runId, f.sourceRunId));
|
||||
if (kind === "lease") await db.update(nativeRunFinalizations).set({ leaseOwner: "owner", leaseExpiresAt: new Date(Date.now() + 60000) }).where(eq(nativeRunFinalizations.runId, f.sourceRunId));
|
||||
if (kind === "process") await db.update(heartbeatRuns).set({ processPid: process.pid }).where(eq(heartbeatRuns.id, f.sourceRunId));
|
||||
if (kind === "launch" || kind === "provider") await db.insert(heartbeatRunEvents).values({ companyId: f.companyId,
|
||||
agentId: f.agentId, runId: f.sourceRunId, seq: 1,
|
||||
eventType: kind === "launch" ? PROCESS_START_REQUESTED : "provider.event",
|
||||
...(kind === "provider" ? { sourceEventId: "provider-1", sourceInstanceId: "provider", sourceSeq: 1, protocolSchemaVersion: 1, canonicalPayloadHash: "hash" } : {}),
|
||||
});
|
||||
if (kind === "cleanup") await db.update(environmentLeases).set({ status: "pending_cleanup", cleanupStatus: "failed" }).where(eq(environmentLeases.heartbeatRunId, f.sourceRunId));
|
||||
if (kind === "remote") await db.update(environmentLeases).set({ provider: "daytona", providerLeaseId: "unverified" }).where(eq(environmentLeases.heartbeatRunId, f.sourceRunId));
|
||||
if (kind === "preparing") await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: new Date(Date.now() + 60000) }).where(eq(heartbeatRuns.id, f.sourceRunId));
|
||||
if (kind === "closed") await db.update(issues).set({ status: "done" }).where(eq(issues.id, f.issueId));
|
||||
if (kind === "reassigned") await db.update(issues).set({ assigneeAgentId: null }).where(eq(issues.id, f.issueId));
|
||||
expect(await admit(f)).toBeNull();
|
||||
expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId)))[0].phase).toBe("observed");
|
||||
await db.delete(environmentLeases).where(eq(environmentLeases.heartbeatRunId, f.sourceRunId));
|
||||
},
|
||||
);
|
||||
|
||||
it("preserves local stop proof after process metadata is cleared and invalidates it on another launch", async () => {
|
||||
const f = await seed();
|
||||
const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId));
|
||||
|
|
@ -85,8 +151,8 @@ const support = await getEmbeddedPostgresTestSupport();
|
|||
expect(await hasNativeLocalProcessStop(db, f.companyId, source.id)).toBe(false);
|
||||
});
|
||||
|
||||
it("resumes saved local messages after restart exactly once and keeps the same wait receipt while blocked", async () => {
|
||||
const f = await seed();
|
||||
it.each(["stopped_process", "cancelled_startup"])("resumes saved local messages after restart exactly once: %s", async kind => {
|
||||
const f = kind === "cancelled_startup" ? await seedCancelledStartup() : await seed();
|
||||
await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" });
|
||||
await db.update(heartbeatRuns).set({ processPid: process.pid }).where(eq(heartbeatRuns.id, f.sourceRunId));
|
||||
// A prior cancelled admission is also held, but cannot select the native
|
||||
|
|
@ -105,7 +171,7 @@ const support = await getEmbeddedPostgresTestSupport();
|
|||
requestedByActorType: "user", requestedByActorId: "board", payload: { issueId: f.issueId, commentId: f.commentId },
|
||||
contextSnapshot: { issueId: f.issueId, wakeCommentId: f.commentId } });
|
||||
const [waiting] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId));
|
||||
expect(waiting.payload?.executionWait).toMatchObject({ reason: "process_running" });
|
||||
expect(waiting.payload?.executionWait).toMatchObject({ reason: kind === "cancelled_startup" ? "controller_settling" : "process_running" });
|
||||
const makeDue = () => db.update(agentWakeupRequests).set({ updatedAt: new Date(0) }).where(eq(agentWakeupRequests.id, waiting.id));
|
||||
await makeDue();
|
||||
await heartbeatService(db).resumeExecutionWaitComments();
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
import { isCancelledNativeStartup } from "./cancelled-native-startup.js";
|
||||
import { hasNativeLocalProcessStop } from "./native-local-process-stop.js";
|
||||
import { completeTerminatedRemoteNativeSessionCleanup } from "../vendor/paperclip-runner/index.js";
|
||||
import { hasRemoteTerminationReceipt, remoteLeaseCleanupScope } from "./remote-execution-termination.js";
|
||||
|
|
@ -74,11 +75,12 @@ export async function admitExplicitNativeContinuation(input: {
|
|||
if (pendingInteraction || pendingApproval) return blocked("decision_pending", "A pending approval or question must be resolved before this message can start.");
|
||||
|
||||
const sources: Run[] = [];
|
||||
const cancelledStartupIds = new Set<string>();
|
||||
for (const action of actions) {
|
||||
const runId = action.evidence.runId ?? action.evidence.sourceRunId;
|
||||
if (typeof runId !== "string") return blocked("source_missing", "The stopped run could not be identified. Your message is saved.");
|
||||
// Text comparison keeps malformed historical evidence a hold, not a UUID cast error.
|
||||
const [run] = await db.select().from(heartbeatRuns).where(and(
|
||||
let [run] = await db.select().from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, companyId), sql`${heartbeatRuns.id}::text = ${runId}`,
|
||||
));
|
||||
if (!run || run.agentId !== agentId || !terminal.includes(run.status) ||
|
||||
|
|
@ -101,12 +103,25 @@ export async function admitExplicitNativeContinuation(input: {
|
|||
// For pre-upgrade rows without adapter evidence, only a new explicit user
|
||||
// turn is allowed, after the termination proofs below. This does not infer
|
||||
// an old adapter type, certify old outcomes, or authorize automatic replay.
|
||||
if (run.runtimeMode !== "native" && !unusedAdmission && !legacyUserTurn) return null;
|
||||
const [coordinator] = await db.select().from(nativeRunFinalizations).where(and(
|
||||
eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, run.id),
|
||||
)).for("update");
|
||||
if (coordinator && (coordinator.phase !== "terminal_failure" || coordinator.leaseOwner ||
|
||||
coordinator.resultId || coordinator.failureDetail?.successorRunId)) return blocked("controller_settling", "Waiting for the previous run to finish recovery. Your message will start automatically.");
|
||||
// Same lock order as the native claim. Re-read the run while holding both
|
||||
// locks before accepting the never-claimed startup proof.
|
||||
const [lockedRun] = await db.select().from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, run.id),
|
||||
)).for("update");
|
||||
if (!lockedRun || lockedRun.status !== run.status || lockedRun.agentId !== run.agentId ||
|
||||
lockedRun.finishedAt?.getTime() !== run.finishedAt.getTime()) return null;
|
||||
run = lockedRun;
|
||||
const cancelledStartup = await isCancelledNativeStartup(db, run, coordinator);
|
||||
if (cancelledStartup) cancelledStartupIds.add(run.id);
|
||||
if (run.runtimeMode !== "native" && !unusedAdmission && !legacyUserTurn && !cancelledStartup) return null;
|
||||
if (!cancelledStartup && coordinator && (coordinator.phase !== "terminal_failure" || coordinator.leaseOwner ||
|
||||
coordinator.resultId || coordinator.failureDetail?.successorRunId)) return blocked("controller_settling",
|
||||
run.status === "cancelled" && !coordinator.leaseOwner
|
||||
? "The cancelled run still needs verified cleanup. Your message is saved. Inspect the run and its environment for details."
|
||||
: "Waiting for the previous run to finish recovery. Your message will start automatically.");
|
||||
const leases = await db.select()
|
||||
.from(environmentLeases).where(and(
|
||||
eq(environmentLeases.companyId, companyId), eq(environmentLeases.heartbeatRunId, run.id),
|
||||
|
|
@ -120,7 +135,7 @@ export async function admitExplicitNativeContinuation(input: {
|
|||
}))) return null;
|
||||
} else {
|
||||
if (leases.some(lease => !lease.releasedAt || lease.cleanupStatus === "failed")) return blocked("local_cleanup", "Waiting for the previous environment to finish cleanup. Your message will start automatically.");
|
||||
if (!unusedAdmission) {
|
||||
if (!unusedAdmission && !cancelledStartup) {
|
||||
// A missing process identity is not evidence that a provider exited.
|
||||
if (!run.processPid && !run.processGroupId &&
|
||||
!await hasNativeLocalProcessStop(db, companyId, run.id)) return blocked("process_identity_missing", "The previous run has no verified stop record. Paperclip cannot start this message yet.");
|
||||
|
|
@ -148,6 +163,16 @@ export async function admitExplicitNativeContinuation(input: {
|
|||
if (input.dryRun) return { previousRunId: previous.id, commentId, ...(retry ? { failedRunId: input.failedRunId! } : {}) };
|
||||
const authorization = { actorId, commentId, ...(retry ? { failedRunId: input.failedRunId } : {}), runId: input.successorRunId,
|
||||
previousRunId: previous.id, recordedAt: new Date().toISOString() };
|
||||
for (const runId of cancelledStartupIds) {
|
||||
await db.update(nativeRunFinalizations).set({
|
||||
phase: "terminal_failure", failureCode: "native_startup_cancelled", nextAttemptAt: null,
|
||||
controlDeadlineAt: null, updatedAt: new Date(),
|
||||
}).where(and(eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, runId)));
|
||||
await db.update(heartbeatRuns).set({
|
||||
...(nativeSources.some(run => run.id === runId) ? { nativePhase: "terminal_failure", nativePhaseUpdatedAt: new Date() } : {}),
|
||||
executionControlDeadlineAt: null,
|
||||
}).where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, runId)));
|
||||
}
|
||||
if (nativeSources.length) await db.update(nativeRunFinalizations).set({
|
||||
failureDetail: sql`coalesce(${nativeRunFinalizations.failureDetail}, '{}'::jsonb) || ${JSON.stringify({ replacementDenied: "explicit_user_continuation" })}::jsonb`,
|
||||
updatedAt: new Date(),
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
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 { legacyControllerBootId, legacyControllerClaim, hasLiveLegacyController, revokeExpiredLegacyController, watchLegacyControllerLease } from "./legacy-controller-lease.js";
|
||||
import { legacyControllerBootId, legacyControllerClaim, renewLegacyControllerLease, hasLiveLegacyController, revokeExpiredLegacyController, watchLegacyControllerLease } from "./legacy-controller-lease.js";
|
||||
import { completeTerminatedRemoteNativeSessionCleanup } from "../vendor/paperclip-runner/index.js";
|
||||
import { remoteExecutionHasStopped, remoteTerminationReceipt, stoppedRemoteCleanupScopes } from "./remote-execution-termination.js";
|
||||
import { applyConnectorSkills, prepareConnectorSkillDelivery, resolveConnectorAssignments } from "./connector-runtime.js";
|
||||
|
|
@ -9044,6 +9044,8 @@ export type HeartbeatEnvironmentRuntime = ReturnType<
|
|||
>;
|
||||
|
||||
export interface HeartbeatServiceOptions {
|
||||
/** Test seam before the atomic native runtime handoff. */
|
||||
beforeNativeRuntimeSelection?: (runId: string) => Promise<void>;
|
||||
/** Test seam immediately before the durable chat-control admission check. */
|
||||
beforeChatControlRecoveryCheck?: (input: {
|
||||
runId: string;
|
||||
|
|
@ -10021,7 +10023,9 @@ export function heartbeatService(
|
|||
|
||||
async function resumeRemoteStopComments(run: typeof heartbeatRuns.$inferSelect, requestId?: string) {
|
||||
if (!isHeartbeatRunTerminalStatus(run.status) || adapterExecutionControls.has(run.id)) return;
|
||||
if (run.runtimeMode !== "native" && !(await remoteExecutionHasStopped(db, run.companyId, run.id))) return;
|
||||
if (run.runtimeMode !== "native" &&
|
||||
parseObject(run.resultJson?.startupCancellation).beforeNativeSelection !== true &&
|
||||
!(await remoteExecutionHasStopped(db, run.companyId, run.id))) return;
|
||||
const issueId = run.nativeIssueId ?? (typeof run.contextSnapshot?.issueId === "string" ? run.contextSnapshot.issueId : null);
|
||||
if (!issueId) return;
|
||||
const legacyContinuation = run.runtimeMode === "legacy" &&
|
||||
|
|
@ -19367,6 +19371,7 @@ export function heartbeatService(
|
|||
Parameters<typeof cleanupGitHubOperationLaunchers>[0] | null = null;
|
||||
let nativeSessionResumeScheduled = false;
|
||||
let nativeOwnershipHeld = false;
|
||||
let nativeDispatchStarted = false;
|
||||
let nativeWorkspaceFinalizeScheduled = false;
|
||||
let nativeWorkspaceSync: Awaited<
|
||||
ReturnType<typeof prepareNativeWorkspaceSync>
|
||||
|
|
@ -22654,7 +22659,8 @@ export function heartbeatService(
|
|||
"destroy_after_turn"
|
||||
? "destroy"
|
||||
: undefined;
|
||||
await db.transaction(async (tx) => {
|
||||
await options.beforeNativeRuntimeSelection?.(run.id);
|
||||
const nativeSelected = await db.transaction(async (tx) => {
|
||||
const lockedRun = await tx
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
|
|
@ -22663,6 +22669,14 @@ export function heartbeatService(
|
|||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!lockedRun) throw new Error("native_runtime_run_missing");
|
||||
// Cancellation and runtime selection serialize on this row. A
|
||||
// stopped preparation must never create a new native coordinator.
|
||||
if (lockedRun.status !== "running" || lockedRun.resultJson?.startupCancellation) return false;
|
||||
if (lockedRun.runtimeMode === "legacy" && lockedRun.controllerBootId &&
|
||||
!(await renewLegacyControllerLease(tx as unknown as Db, lockedRun))) {
|
||||
nativeOwnershipHeld = true;
|
||||
return false;
|
||||
}
|
||||
if (
|
||||
lockedRun.runtimeModeResolvedAt &&
|
||||
lockedRun.runtimeMode !== "native"
|
||||
|
|
@ -22770,7 +22784,9 @@ export function heartbeatService(
|
|||
phase: "observed",
|
||||
})
|
||||
.onConflictDoNothing();
|
||||
return true;
|
||||
});
|
||||
if (!nativeSelected) return;
|
||||
controllerLease.stop();
|
||||
nativeWorkspaceSync = await prepareNativeWorkspaceSync({
|
||||
db,
|
||||
|
|
@ -23303,6 +23319,7 @@ export function heartbeatService(
|
|||
}),
|
||||
);
|
||||
if (!guardedDispatch.dispatched) return;
|
||||
nativeDispatchStarted = true;
|
||||
adapterResult = await guardedDispatch.resultPromise;
|
||||
} finally {
|
||||
await nativeGitHubBridge?.stop();
|
||||
|
|
@ -25063,6 +25080,17 @@ export function heartbeatService(
|
|||
});
|
||||
}
|
||||
}
|
||||
if (latestRun?.status === "cancelled" && !nativeDispatchStarted && !nativeOwnershipHeld &&
|
||||
(latestRun.runtimeMode === "native" ||
|
||||
parseObject(latestRun.resultJson?.startupCancellation).beforeNativeSelection === true)) {
|
||||
// This executor has finished preparation and lease cleanup without
|
||||
// handing off to native execution. Keep a durable receipt for admission
|
||||
// after a restart; cleanup receipts are independently rechecked there.
|
||||
await db.update(heartbeatRuns).set({
|
||||
resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) ||
|
||||
${JSON.stringify({ startupPreparationSettledAt: new Date().toISOString() })}::jsonb`,
|
||||
}).where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.status, "cancelled")));
|
||||
}
|
||||
// Interrupting a queued message explicitly authorizes the pending queue.
|
||||
// Retry its normal promotion after leases and adapter cleanup have settled;
|
||||
// the earlier terminal write can still have an execution blocker here.
|
||||
|
|
@ -27619,7 +27647,7 @@ export function heartbeatService(
|
|||
reason = "Cancelled by control plane",
|
||||
options: CancelRunOptions = {},
|
||||
) {
|
||||
const run = await getRun(runId);
|
||||
let run = await getRun(runId);
|
||||
if (!run) throw notFound("Heartbeat run not found");
|
||||
const pendingNativeRetry =
|
||||
run.runtimeMode === "native" && run.status === "failed"
|
||||
|
|
@ -27644,16 +27672,6 @@ export function heartbeatService(
|
|||
return run;
|
||||
const agent = await getAgent(run.agentId);
|
||||
const errorCode = options.errorCode ?? "cancelled";
|
||||
const resultJson = agent
|
||||
? {
|
||||
...mergeRunStopMetadataForAgent(agent, "cancelled", {
|
||||
resultJson: parseObject(run.resultJson),
|
||||
errorCode,
|
||||
errorMessage: reason,
|
||||
}),
|
||||
...(options.resultJson ?? {}),
|
||||
}
|
||||
: options.resultJson;
|
||||
|
||||
const pendingProcessCancellation = processRunCancellationSettlements.get(
|
||||
run.id,
|
||||
|
|
@ -27670,6 +27688,39 @@ export function heartbeatService(
|
|||
? captureAdapterStopOwnership(run.id)
|
||||
: undefined;
|
||||
const control = stopOwnership?.control;
|
||||
// Capture the existing adapter owner before waiting on the run lock. Then
|
||||
// atomically fence preparation and refresh the selected runtime, so Stop
|
||||
// cannot miss a native handoff that won after its first read.
|
||||
// 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)) {
|
||||
const [fenced] = await db.update(heartbeatRuns).set({
|
||||
resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::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)
|
||||
))`,
|
||||
}).where(and(eq(heartbeatRuns.id, runId), inArray(heartbeatRuns.status,
|
||||
pendingNativeRetry ? [...CANCELLABLE_HEARTBEAT_RUN_STATUSES, "failed"] : [...CANCELLABLE_HEARTBEAT_RUN_STATUSES],
|
||||
))).returning();
|
||||
if (!fenced) return getRun(runId);
|
||||
run = fenced;
|
||||
}
|
||||
const resultJson = agent
|
||||
? {
|
||||
...mergeRunStopMetadataForAgent(agent, "cancelled", {
|
||||
resultJson: parseObject(run.resultJson),
|
||||
errorCode,
|
||||
errorMessage: reason,
|
||||
}),
|
||||
...(options.resultJson ?? {}),
|
||||
}
|
||||
: options.resultJson;
|
||||
|
||||
try {
|
||||
let releaseProcessCancellation: (() => void) | undefined;
|
||||
const processCancellationSettlement =
|
||||
|
|
|
|||
|
|
@ -4146,6 +4146,7 @@ function leaseDb(
|
|||
runResultJson: Record<string, unknown> = {},
|
||||
updates: Array<{ table: unknown; values: Record<string, unknown> }> = [],
|
||||
runnerProfileJson: Record<string, unknown> = {},
|
||||
runStatus = "running",
|
||||
): Db {
|
||||
const coordinator: LeaseCoordinator = {
|
||||
runId: boundExecution.binding.runId,
|
||||
|
|
@ -4189,6 +4190,7 @@ function leaseDb(
|
|||
resultJson: runResultJson,
|
||||
runnerProfileJson,
|
||||
runtimeMode: "native",
|
||||
status: runStatus,
|
||||
},
|
||||
]
|
||||
: table === issues
|
||||
|
|
@ -6527,6 +6529,28 @@ describe("native process ownership", () => {
|
|||
);
|
||||
});
|
||||
|
||||
it.each(["cancelled", "succeeded", "interrupted", "timed_out", "failed"])(
|
||||
"refuses native provider claims after the run became %s", async status => {
|
||||
const updates: Array<{ table: unknown; values: Record<string, unknown> }> = [];
|
||||
state.createBackend.mockClear();
|
||||
await expect(executePaperclipNativeSession({
|
||||
db: leaseDb(execution, {}, {}, updates, {}, status), execution, runnerInstanceId: "late-startup",
|
||||
})).rejects.toThrow();
|
||||
expect(state.createBackend).not.toHaveBeenCalled();
|
||||
expect(updates.some(update => update.table === nativeRunFinalizations)).toBe(false);
|
||||
expect(updates.some(update => update.values.eventType === "native.process_start_requested")).toBe(false);
|
||||
},
|
||||
);
|
||||
|
||||
it("fences a cancellation request before its terminal status commits", async () => {
|
||||
state.createBackend.mockClear();
|
||||
await expect(executePaperclipNativeSession({
|
||||
db: leaseDb(execution, {}, { startupCancellation: { requestedAt: new Date().toISOString() } }),
|
||||
execution, runnerInstanceId: "cancel-requested",
|
||||
})).rejects.toThrow();
|
||||
expect(state.createBackend).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("forwards the app-server PID and process group through the production backend seam", async () => {
|
||||
const processMetadata = {
|
||||
pid: 42_001,
|
||||
|
|
|
|||
|
|
@ -6965,6 +6965,7 @@ async function executePaperclipNativeSessionWithinScope(
|
|||
nativeIssueId: heartbeatRuns.nativeIssueId,
|
||||
resultJson: heartbeatRuns.resultJson,
|
||||
runtimeMode: heartbeatRuns.runtimeMode,
|
||||
status: heartbeatRuns.status,
|
||||
})
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.id, input.execution.binding.runId))
|
||||
|
|
@ -6980,6 +6981,12 @@ async function executePaperclipNativeSessionWithinScope(
|
|||
) {
|
||||
throw new Error("native_execution_binding_changed");
|
||||
}
|
||||
// A cancellation can win after heartbeat dispatch admission but
|
||||
// before this claim. Never revive a terminal run or a settled startup.
|
||||
if (boundRun.status !== "running" || boundRun.resultJson?.startupCancellation ||
|
||||
coordinator.phase === "terminal_failure") {
|
||||
throw new NativeCancellationPendingRecoveryError();
|
||||
}
|
||||
const cancellationIntent = record(
|
||||
record(boundRun.resultJson).nativeCancellation,
|
||||
);
|
||||
|
|
|
|||
Loading…
Reference in New Issue