mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-04 02:32:07 +01:00
fix(runner): preserve cleanup ownership across startup failures (#18658)
## Thinking Path > - Paperclip controls which run owns an agent task. > - A plugin can continue accepted work after Core stops waiting. > - A late task-creation reply can therefore arrive after startup cancellation. > - Core must keep that work owned until its resource identity and cleanup outcome are known. > - This PR supplies the cleanup and recovery support before shared startup is enabled. > - The dependent activation PR can then reuse transport data within a bounded startup context. ## Linked Issues or Issue Description Refs #18405. This is the independently mergeable prerequisite for that stacked activation PR. **What existing behavior does this improve?** Cleanup of explicitly recorded startup work that outlives its caller. **Current behavior** Caller timeout, worker exit, or failed cleanup-intent persistence can leave work without a complete recovery path. **Proposed behavior** Retain the cleanup obligation, use existing Stop and recovery queues, and release ownership only after the exact task and accepted work have stopped. Ordinary calls keep their existing startup payloads and budgets. This PR does not create startup contexts or advertise transport reuse. **Reason and benefit** Separate the failure-handling contract from activation so each can be reviewed on its own. This support PR has **no standalone TTFT speedup**. ## What Changed - Add the startup cleanup marker, exact ownership checks, verified completion, and acknowledgement helpers. - Connect recorded cleanup obligations to existing Stop, deadline reconciliation, restart recovery, and retry scheduling. - Retain typed cleanup responsibility when a database write fails. Keep the nonterminal run from being treated as ordinary completed cleanup. - Observe late acquisition results and save resource identities before retiring them. Retain real host writes and received results across worker exit or persistence failure. - Queue abandoned replacement tasks through the existing creation journal and cleanup machinery. Require the replacement's own Stop proof. - Accept an explicit startup context in the internal bridge for lifetime tracking. Automatic context creation, marker production, capability advertisement, transport capture, and executor activation belong to #18405. - Add PostgreSQL and worker IPC regressions, including legacy payload assertions. ## Verification Initial split source `8fb482d8be9e70101b0cfdd0830f41cf5b8e3cf4` had tree `95508bd1859e991670cad8afd4de8d2167009ac5`. **190 distinct targeted tests pass**, along with server typechecking. The selected checks cover real PostgreSQL cleanup/ownership, worker IPC, the bridge and creation journal, ordinary/distributed Stop, restart and deadline recovery, idle-sleep behavior, and the unchanged legacy warm-task suite. Independent recovery and worker reviews found no blocking issue. The first correction, `b6342d42042601b376da38dc8e1dba9ca80e9539`, fixes the replacement cleanup finding with the existing startup-cleanup marker, scoped to contextual abandoned task creations. The replacement was never prepared, so preparation-stop proof is not required. No dispatch is fabricated and no normal-startup query is added. The actual queue/sweep/driver/receipt/cancellation regression fails before the fix and passes after it: void, running or foreign-task status retains ownership; exact-task stopped status permits completion even when Release returns void. **64 affected targeted cases and server typechecking pass**. Independent review found no blocker. Current source is `f53af72af76f263096c3c2e788506e6faef9474f`. It fixes both recovery findings from the prior head's Apex 3/5: orphan recovery now preserves pending or completed-but-not-cancelled cleanup through its existing final-write fence, and exact late identities can reach verified cleanup after reconciliation or restart. The regression selection went from four failing assertions to six passing cases. Affected selections cover 217 distinct cases; the exact final six-case regression and server typecheck pass. Independent reviewers found no blocker in either fix. Current-head standard review is 5/5 and all CI checks pass (54 successes, three expected skips, plus Snyk). Apex also completed at **5/5** on this exact head, with zero unresolved threads. The requested independent availability audit is complete on this head. It found an activation-dependent operability gap: an acquisition response that is never received can leave an unknown creation identity with no supported task inspection/reconciliation path, while cleanup keeps the run owner reserved. This does not occur in ordinary lower-only calls. The contract and recovery policy need follow-up before activation; the external review scores do not resolve that availability tradeoff. The split preserves the accepted implementation at `21db7b3851` after its clean merge with frozen master `946735d03a55c1fb4cc5434d6bf1d87742daa9df`. The initial dependent activation PR restored exact aggregate tree `f68c6895f87c7b93ae03e6324d955d2659a99dce`. Both subsequent cleanup corrections are integrated into upper head `27978f294a02da3992585c7f4daa73cd8fcf99f8`; its current aggregate tree is `a0b66c33f5e22745f034f45d24d213f5c210cf16`. Lower-only payload tests cover the absence of capability advertisement and transport projection. Only targeted local tests and typechecking are used. Full regression coverage runs in GitHub CI. No deployment or ROM qualification is part of this change. The lower audit covered connected worker, creation-journal, Stop, restart, deadline and orphan-recovery paths. A new isolated real-PostgreSQL characterization reproduced the unknown-ID hold through actual recovery, cleanup verification and run-control cancellation. It models a lost acquisition response; it is not a live provider or complete IPC test. No source edits were made by the audit. The upper/composition audit remains required after its review cycle. ## Risks - Recording a replacement result now commits with the existing cleanup notification. This conditional replacement path adds transaction/notification responses; common retained startup is unchanged. - This adds cleanup and recovery state handling. A missing receipt or permanently inaccessible old backend can keep ownership held indefinitely. Temporary outages retry; an ordinary retry is not evidence that a deleted or replaced backend stopped. - A caller timeout does not undo work already accepted by a plugin. Unknown resource creation remains unresolved until its identity/outcome is established. Current task creation has no supported inspect-by-operation or operator reconciliation API for a permanently lost identity; repeated Stop alone cannot resolve it. - The support code understands records produced by the dependent activation PR. It does not enable the shared startup policy by itself. - Existing legacy callers must retain their behavior; context-only automatic replacement cleanup remains explicitly scoped. ## Model Used OpenAI Codex, GPT-6 family. The exact model ID and context window are not exposed in this session. Tool-assisted implementation, targeted testing, and independent subagent review. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used, with available version and capability details - [x] I have checked ROADMAP.md and confirmed this is the requested split of existing work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have described the issue with the required fields - [x] I have not referenced internal issues or private links - [x] My branch name describes the change and contains no internal ticket ID - [x] Targeted local tests pass on this source - [x] I have added or updated relevant tests - [x] I have updated relevant documentation - [x] I have documented the risks - [x] All current-head GitHub CI gates are green - [x] Current-head standard Greptile and Apex are 5/5 with no unresolved findings - [x] I will address all review comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Nicky LeachandPaperclip authored and GitHub committed
1 parent
b4315943e8
commit
a9b18e1cbe
24 files changed
+1829
-106
No files matched your search
@@ -32,8 +32,8 @@ const invocations = [];
|
||||
const send = (id, result) => process.stdout.write(JSON.stringify({jsonrpc:"2.0",id,result})+"\\n");
|
||||
readline.createInterface({input:process.stdin,crlfDelay:Infinity}).on("line", line => {
|
||||
const message = JSON.parse(line);
|
||||
if (message.method === "initialize") send(message.id, {ok:true,supportedMethods:["environmentTask","configChanged","health"]});
|
||||
else if (message.method === "environmentTask" || message.method === "environmentDestroyLease") {
|
||||
if (message.method === "initialize") send(message.id, {ok:true,supportedMethods:["environmentTask","environmentAcquireLease","configChanged","health"]});
|
||||
else if (message.method === "environmentTask" || message.method === "environmentDestroyLease" || message.method === "environmentAcquireLease") {
|
||||
held.push(message.id);
|
||||
invocations.push(message.paperclipInvocation?.id);
|
||||
if (message.params.config.hostWrite) process.stdout.write(JSON.stringify({jsonrpc:"2.0",id:"host-"+message.id,method:"companies.get",params:{companyId:"company"},paperclipInvocationId:message.paperclipInvocation.id})+"\\n");
|
||||
@@ -43,7 +43,7 @@ readline.createInterface({input:process.stdin,crlfDelay:Infinity}).on("line", li
|
||||
else if (message.method === "configChanged") {
|
||||
if (message.params.config.exit) process.exit(1);
|
||||
if (message.params.config.hostAfter) process.stdout.write(JSON.stringify({jsonrpc:"2.0",id:"late-host",method:"companies.get",params:{companyId:"company"},paperclipInvocationId:invocations.at(-1)})+"\\n");
|
||||
if (message.params.config.release) for (const id of held.splice(0)) send(id, {kind:"description",taskId:"task"});
|
||||
if (message.params.config.release) for (const id of held.splice(0)) { send(id, {providerLeaseId:"late-task",kind:"description",taskId:"task"}); if (message.params.config.duplicate) send(id, {providerLeaseId:"late-task",kind:"description",taskId:"task"}); }
|
||||
send(message.id, {});
|
||||
} else if (message.method === "shutdown") {send(message.id, {});setImmediate(()=>process.exit(0));}
|
||||
});
|
||||
@@ -59,6 +59,70 @@ async function startWorker(hostHandlers: Record<string, unknown> = {}) {
|
||||
}
|
||||
|
||||
describe("Core startup responsibility across real worker IPC", () => {
|
||||
it("drops only a dead acquisition reply placeholder while retaining real writes until verified retirement", async () => {
|
||||
let release!: () => void, writing!: () => void;
|
||||
const gate = new Promise<void>(resolve => { release = resolve; }), entered = new Promise<void>(resolve => { writing = resolve; });
|
||||
const manager = await startWorker({ "companies.get": async () => { writing(); await gate; return {}; } });
|
||||
const settled = vi.fn(), observe = vi.fn(async () => {});
|
||||
try {
|
||||
const acquisition = manager.call(manifest.id, "environmentAcquireLease", { companyId: "company", environmentId: "environment",
|
||||
driverKey: "tasks", config: { hostWrite: true }, runId: "run", agentId: "agent", adapterType: "paperclip_runner" },
|
||||
10, undefined, settled, "dead-acquire", observe);
|
||||
const rejected = expect(acquisition).rejects.toThrow("timed out");
|
||||
await entered; await rejected;
|
||||
await manager.stopWorker(manifest.id);
|
||||
let drained = false;
|
||||
const draining = manager.drainStartup!("dead-acquire").then(() => { drained = true; });
|
||||
await new Promise<void>(resolve => setImmediate(resolve));
|
||||
expect(drained).toBe(false);
|
||||
expect(settled).not.toHaveBeenCalled();
|
||||
release(); await draining;
|
||||
expect(observe).not.toHaveBeenCalled();
|
||||
expect(settled).not.toHaveBeenCalled();
|
||||
await manager.retireStartup!("dead-acquire");
|
||||
expect(settled).toHaveBeenCalledOnce();
|
||||
} finally { release(); await manager.stopAll(); }
|
||||
}, 25_000);
|
||||
|
||||
it.each([false, true])("retains a received acquisition reply through failed recording and retirement (worker exits=%s)", async workerExits => {
|
||||
const manager = await startWorker(), settled = vi.fn();
|
||||
let release!: () => void, recording!: () => void;
|
||||
const gate = new Promise<void>(resolve => { release = resolve; }), started = new Promise<void>(resolve => { recording = resolve; });
|
||||
let unavailable = true;
|
||||
const observe = vi.fn(async (result: unknown) => {
|
||||
expect(result).toMatchObject({ providerLeaseId: "late-task" });
|
||||
if (unavailable) throw new Error("Database unavailable");
|
||||
recording(); await gate;
|
||||
});
|
||||
try {
|
||||
const acquire = manager.call(manifest.id, "environmentAcquireLease", { companyId: "company", environmentId: "environment",
|
||||
driverKey: "tasks", config: {}, runId: "run", agentId: "agent", adapterType: "paperclip_runner" }, 10, undefined, settled, "late-acquire", observe);
|
||||
await expect(acquire).rejects.toThrow("timed out");
|
||||
await manager.call(manifest.id, "configChanged", { config: { release: true, duplicate: true } });
|
||||
await vi.waitFor(() => expect(observe).toHaveBeenCalledOnce());
|
||||
await new Promise<void>(resolve => setImmediate(resolve));
|
||||
expect(settled).not.toHaveBeenCalled();
|
||||
if (workerExits) await manager.stopWorker(manifest.id);
|
||||
unavailable = false;
|
||||
vi.useFakeTimers();
|
||||
const failed = manager.retireStartup!("late-acquire").catch(error => error.message);
|
||||
await started;
|
||||
expect(observe).toHaveBeenCalledTimes(2);
|
||||
await vi.advanceTimersByTimeAsync(15_000);
|
||||
expect(await failed).toBe("Task cleanup is still pending");
|
||||
expect(settled).not.toHaveBeenCalled();
|
||||
vi.useRealTimers();
|
||||
let retired = false;
|
||||
const retiring = manager.retireStartup!("late-acquire").then(() => { retired = true; });
|
||||
await new Promise<void>(resolve => setImmediate(resolve));
|
||||
expect(retired).toBe(false);
|
||||
release(); await retiring;
|
||||
expect(settled).toHaveBeenCalledOnce();
|
||||
await manager.retireStartup!("late-acquire");
|
||||
expect(observe).toHaveBeenCalledTimes(2);
|
||||
} finally { vi.useRealTimers(); release(); await manager.stopAll(); }
|
||||
});
|
||||
|
||||
it.each(["stop", "replace", "stopAll"])("keeps stopped startup work reachable through %s until verified retirement", async mode => {
|
||||
let release!: () => void, entered!: () => void;
|
||||
const gate = new Promise<void>(resolve => { release = resolve; }), started = new Promise<void>(resolve => { entered = resolve; });
|
||||
|
||||
@@ -38,6 +38,7 @@ import { isNewTerminalRunFailure, reportRunFailure } from "../run-failure-report
|
||||
import { isNativeRunnerOwnershipHeld } from "./native-runner-ownership.js";
|
||||
import { nativeRestartInterruptedTurnId, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js";
|
||||
import { retireSupersededNativeRetry } from "./native-superseded-retry.js";
|
||||
import { requestNativeStartupCleanup, readNativeStartupContext, promoteNativeStartupCleanup } from "./native-startup-cleanup.js";
|
||||
|
||||
export type NativeControllerIdentity = {
|
||||
bootId: string;
|
||||
@@ -623,22 +624,41 @@ export async function claimNativeRestartRecoveries(input: {
|
||||
} as const;
|
||||
}
|
||||
|
||||
const startupContext = readNativeStartupContext(row.run, row.coordinator);
|
||||
const startupRetryAt = startupContext ? new Date(Math.max(now.getTime() + 30_000,
|
||||
startupContext.deadlineAtMs, row.coordinator.leaseExpiresAt?.getTime() ?? 0)) : undefined;
|
||||
if (input.liveRunExecutions?.has(row.run.id)) {
|
||||
return {
|
||||
kind: "awaiting_evidence",
|
||||
runId: row.run.id,
|
||||
reason: "current_controller_preparing",
|
||||
...(startupRetryAt ? { retryAt: startupRetryAt } : {}),
|
||||
} as const;
|
||||
}
|
||||
if (startupContext) {
|
||||
if (await promoteNativeStartupCleanup(tx as unknown as Db, row.run, row.coordinator, now)) {
|
||||
return { kind: "blocked", runId: row.run.id, reason: "native_startup_cleanup_pending" } as const;
|
||||
}
|
||||
return { kind: "awaiting_evidence", runId: row.run.id, reason: "native_startup_cleanup_pending", retryAt: startupRetryAt } as const;
|
||||
}
|
||||
|
||||
const profile = row.run.runnerProfileJson ?? {};
|
||||
const storedRemoteWorkspace = remoteWorkspaceRunReferenceSchema.safeParse(profile.remoteWorkspace);
|
||||
const remoteWorkspace = readNativeWorkspaceSyncReference(profile.nativeWorkspaceSync)
|
||||
?? (storedRemoteWorkspace.success ? storedRemoteWorkspace.data : null);
|
||||
const [pluginTask] = await tx.select({ id: environmentLeases.id }).from(environmentLeases).where(and(
|
||||
const [pluginTask] = await tx.select({ id: environmentLeases.id, metadata: environmentLeases.metadata }).from(environmentLeases).where(and(
|
||||
eq(environmentLeases.companyId, row.run.companyId), eq(environmentLeases.heartbeatRunId, row.run.id),
|
||||
sql`${environmentLeases.metadata}->>'runnerTaskLease' = 'true'`,
|
||||
)).orderBy(desc(environmentLeases.createdAt)).limit(1);
|
||||
const startup = pluginTask?.metadata?.runnerTaskStartup as { contextId?: unknown; deadlineAtMs?: unknown; runId?: unknown } | undefined;
|
||||
if (startup?.runId === row.run.id && typeof startup.contextId === "string" &&
|
||||
Number.isSafeInteger(startup.deadlineAtMs) && Number(startup.deadlineAtMs) <= now.getTime() &&
|
||||
!row.run.sessionIdAfter && row.coordinator.leaseOwner && row.coordinator.leaseExpiresAt && row.coordinator.leaseExpiresAt <= now) {
|
||||
await requestNativeStartupCleanup(tx as unknown as Db, { companyId: row.run.companyId, agentId: row.run.agentId,
|
||||
runId: row.run.id, contextId: startup.contextId, deadlineAtMs: Number(startup.deadlineAtMs),
|
||||
leaseOwner: row.coordinator.leaseOwner, requiresPreparationStop: true, leaseId: pluginTask!.id }, { publish: false });
|
||||
return { kind: "blocked", runId: row.run.id, reason: "native_startup_cleanup_pending" } as const;
|
||||
}
|
||||
const remoteSandbox = profile.nativeWorkspaceSync != null || profile.remoteWorkspace != null;
|
||||
const remoteProcess = remoteSandbox || !!pluginTask;
|
||||
const durableAuthority = (remoteWorkspace || pluginTask) && await hasNativeRunnerCheckpoint(tx, {
|
||||
|
||||
@@ -0,0 +1,208 @@
|
||||
import { and, eq, isNull, sql } from "drizzle-orm";
|
||||
import { environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, sandboxCreationOperations, type Db } from "@paperclipai/db";
|
||||
import { conflict } from "../../errors.js";
|
||||
import { persistActivity, publishActivity } from "../activity-log.js";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
|
||||
import { hasRemoteTerminationReceipt } from "../remote-execution-termination.js";
|
||||
import { isNativeRunnerOwnershipHeld } from "./native-runner-ownership.js";
|
||||
|
||||
export const NATIVE_STARTUP_CLEANUP_PENDING = "native_startup_cleanup_pending";
|
||||
|
||||
export class NativeStartupCleanupPendingError extends Error {
|
||||
constructor(readonly contextId: string, readonly deadlineAtMs: number, readonly leaseOwner: string,
|
||||
readonly requiresPreparationStop: boolean, readonly leaseId?: string) {
|
||||
super(NATIVE_STARTUP_CLEANUP_PENDING);
|
||||
this.name = "NativeStartupCleanupPendingError";
|
||||
}
|
||||
}
|
||||
|
||||
type Run = typeof heartbeatRuns.$inferSelect;
|
||||
export interface NativeStartupCleanup {
|
||||
schema: "paperclip.native-startup-cleanup.v1";
|
||||
companyId: string; agentId: string; runId: string;
|
||||
contextId: string; deadlineAtMs: number; leaseOwner: string;
|
||||
controllerGeneration: number; attempt: number; leaseId: string; taskId: string;
|
||||
requiresPreparationStop: boolean; state: "pending" | "complete"; auditId: string;
|
||||
}
|
||||
|
||||
export function readNativeStartupCleanup(run: Pick<Run, "id" | "companyId" | "agentId" | "resultJson">): NativeStartupCleanup | null {
|
||||
const marker = run.resultJson?.nativeStartupCleanup as NativeStartupCleanup | undefined;
|
||||
return marker?.schema === "paperclip.native-startup-cleanup.v1" && marker.companyId === run.companyId &&
|
||||
marker.runId === run.id && marker.agentId === run.agentId && ["contextId", "leaseOwner", "leaseId", "taskId", "auditId"].every(key =>
|
||||
typeof marker[key as keyof NativeStartupCleanup] === "string" && String(marker[key as keyof NativeStartupCleanup]).length > 0) &&
|
||||
Number.isSafeInteger(marker.deadlineAtMs) && Number.isSafeInteger(marker.controllerGeneration) && marker.controllerGeneration > 0 &&
|
||||
Number.isSafeInteger(marker.attempt) && marker.attempt > 0 && typeof marker.requiresPreparationStop === "boolean" &&
|
||||
["pending", "complete"].includes(marker.state) ? marker : null;
|
||||
}
|
||||
|
||||
export function readNativeStartupContext(run: Run,
|
||||
owner: { leaseOwner: string | null; controllerGeneration: number; attempt: number; resultId: string | null }) {
|
||||
const value = run.runnerProfileJson?.nativeStartupContext as { schema?: unknown; companyId?: unknown; agentId?: unknown; runId?: unknown;
|
||||
contextId?: unknown; deadlineAtMs?: unknown; leaseId?: unknown; leaseOwner?: unknown; controllerGeneration?: unknown; attempt?: unknown } | undefined;
|
||||
if (run.runtimeMode !== "native" || run.status !== "running" || run.sessionIdAfter || owner.resultId || isNativeRunnerOwnershipHeld(run) ||
|
||||
!owner.leaseOwner || owner.controllerGeneration < 1 || owner.attempt < 1 || value?.schema !== "paperclip.native-startup-context.v1" ||
|
||||
value.companyId !== run.companyId || value.agentId !== run.agentId || value.runId !== run.id || value.leaseOwner !== owner.leaseOwner ||
|
||||
value.controllerGeneration !== owner.controllerGeneration || value.attempt !== owner.attempt ||
|
||||
typeof value.contextId !== "string" || !value.contextId || typeof value.leaseId !== "string" || !value.leaseId ||
|
||||
!Number.isSafeInteger(value.deadlineAtMs)) return null;
|
||||
return { contextId: value.contextId, leaseId: value.leaseId, deadlineAtMs: Number(value.deadlineAtMs) };
|
||||
}
|
||||
|
||||
export async function clearNativeStartupContext(db: Db, input: {
|
||||
companyId: string; runId: string; contextId: string; leaseOwner: string; controllerGeneration: number;
|
||||
}): Promise<void> {
|
||||
const cleared = await db.update(heartbeatRuns).set({
|
||||
runnerProfileJson: sql`${heartbeatRuns.runnerProfileJson} - 'nativeStartupContext'`,
|
||||
}).where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, input.runId),
|
||||
eq(heartbeatRuns.status, "running"),
|
||||
sql`coalesce(${heartbeatRuns.resultJson}->'nativeStartupCleanup'->>'state', '') != 'pending'`,
|
||||
sql`${heartbeatRuns.runnerProfileJson}->'nativeStartupContext'->>'contextId' = ${input.contextId}`,
|
||||
sql`${heartbeatRuns.runnerProfileJson}->'nativeStartupContext'->>'leaseOwner' = ${input.leaseOwner}`,
|
||||
sql`${heartbeatRuns.runnerProfileJson}->'nativeStartupContext'->>'controllerGeneration' = ${String(input.controllerGeneration)}`,
|
||||
sql`exists (select 1 from ${nativeRunFinalizations} where ${nativeRunFinalizations.runId} = ${input.runId}
|
||||
and ${nativeRunFinalizations.leaseOwner} = ${input.leaseOwner}
|
||||
and ${nativeRunFinalizations.controllerGeneration} = ${input.controllerGeneration})`,
|
||||
)).returning({ id: heartbeatRuns.id });
|
||||
if (cleared.length !== 1) throw new Error("Native startup fallback lost authority");
|
||||
}
|
||||
|
||||
export async function promoteNativeStartupCleanup(db: Db, run: Run,
|
||||
owner: { leaseOwner: string | null; leaseExpiresAt: Date | null; controllerGeneration: number; attempt: number; resultId: string | null },
|
||||
now: Date): Promise<boolean> {
|
||||
const startup = readNativeStartupContext(run, owner);
|
||||
if (!startup || startup.deadlineAtMs > now.getTime() || !owner.leaseOwner || !owner.leaseExpiresAt || owner.leaseExpiresAt > now || isNativeRunnerOwnershipHeld(run)) return false;
|
||||
const [lease] = await db.select().from(environmentLeases).where(and(eq(environmentLeases.id, startup.leaseId),
|
||||
eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id),
|
||||
sql`${environmentLeases.metadata}->>'runnerTaskLease' = 'true'`));
|
||||
if (!lease) return false;
|
||||
const prepared = lease.metadata?.runnerTaskStartup as { contextId?: unknown; runId?: unknown } | undefined;
|
||||
return requestNativeStartupCleanup(db, { companyId: run.companyId, agentId: run.agentId, runId: run.id,
|
||||
...startup, leaseOwner: owner.leaseOwner,
|
||||
requiresPreparationStop: prepared?.contextId === startup.contextId && prepared.runId === run.id }, { publish: false });
|
||||
}
|
||||
|
||||
export async function requestNativeStartupCleanup(db: Db, input: {
|
||||
companyId: string; agentId: string; runId: string; contextId: string; deadlineAtMs: number; leaseOwner: string;
|
||||
requiresPreparationStop: boolean;
|
||||
leaseId?: string;
|
||||
}, options: { publish?: boolean } = {}): Promise<boolean> {
|
||||
let publication: Parameters<typeof publishActivity>[0] | null = null;
|
||||
const [binding] = await db.select({ issueId: nativeRunFinalizations.issueId }).from(nativeRunFinalizations).where(and(
|
||||
eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.runId, input.runId)));
|
||||
if (!binding) return false;
|
||||
const requested = await db.transaction(async tx => {
|
||||
await tx.select({ id: issues.id }).from(issues).where(and(eq(issues.companyId, input.companyId), eq(issues.id, binding.issueId))).for("update");
|
||||
const [coordinator] = await tx.select().from(nativeRunFinalizations).where(and(
|
||||
eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.runId, input.runId))).for("update");
|
||||
const [run] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, input.companyId),
|
||||
eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.agentId, input.agentId))).for("update");
|
||||
if (!run || !coordinator || run.runtimeMode !== "native" || run.status !== "running" || run.sessionIdAfter || coordinator.resultId ||
|
||||
isNativeRunnerOwnershipHeld(run)) return false;
|
||||
const existing = readNativeStartupCleanup(run);
|
||||
if (existing) return existing.contextId === input.contextId;
|
||||
if (coordinator.leaseOwner !== input.leaseOwner) return false;
|
||||
const leases = await tx.select().from(environmentLeases).where(and(eq(environmentLeases.companyId, input.companyId),
|
||||
eq(environmentLeases.heartbeatRunId, input.runId), input.leaseId ? eq(environmentLeases.id, input.leaseId) : undefined,
|
||||
sql`${environmentLeases.metadata}->>'runnerTaskLease' = 'true'`)).for("update");
|
||||
if (leases.length !== 1 || !leases[0]!.providerLeaseId) throw new Error("Native startup cleanup lease unavailable");
|
||||
const lease = leases[0]!;
|
||||
const activity = await persistActivity(tx as unknown as Db, { companyId: input.companyId, actorType: "system",
|
||||
actorId: "native-startup-cleanup", action: "native.startup_cleanup_requested", entityType: "heartbeat_run", entityId: run.id,
|
||||
agentId: run.agentId, runId: run.id, issueId: run.nativeIssueId ?? undefined,
|
||||
details: { contextId: input.contextId, deadlineAtMs: input.deadlineAtMs, leaseId: lease.id,
|
||||
controllerGeneration: coordinator.controllerGeneration, reason: "startup_worker_unsettled" } });
|
||||
if (!activity.activity?.id) throw new Error("Native startup cleanup audit unavailable");
|
||||
publication = activity.publication;
|
||||
const preparation = lease.metadata?.runnerTaskStartup as { contextId?: unknown; runId?: unknown } | undefined;
|
||||
const marker: NativeStartupCleanup = { schema: "paperclip.native-startup-cleanup.v1", companyId: run.companyId, agentId: run.agentId,
|
||||
runId: run.id, contextId: input.contextId, deadlineAtMs: input.deadlineAtMs, leaseOwner: input.leaseOwner,
|
||||
controllerGeneration: coordinator.controllerGeneration, attempt: coordinator.attempt, leaseId: lease.id,
|
||||
taskId: lease.providerLeaseId!, requiresPreparationStop: input.requiresPreparationStop ||
|
||||
(preparation?.contextId === input.contextId && preparation.runId === run.id), state: "pending", auditId: activity.activity.id };
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionFinalization);
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.environmentCleanup);
|
||||
await tx.update(heartbeatRuns).set({ errorCode: NATIVE_STARTUP_CLEANUP_PENDING,
|
||||
error: "Native startup timed out. Cancellation and verified cleanup are pending.",
|
||||
resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({ nativeStartupCleanup: marker,
|
||||
startupCancellation: { ...run.resultJson?.startupCancellation as object, requestedAt: new Date().toISOString() } })}::jsonb`,
|
||||
updatedAt: new Date() }).where(eq(heartbeatRuns.id, run.id));
|
||||
await tx.update(nativeRunFinalizations).set({ phase: "terminal_failure", failureCode: NATIVE_STARTUP_CLEANUP_PENDING,
|
||||
nextAttemptAt: null, controlDeadlineAt: new Date(), recoveryState: "blocked", updatedAt: new Date() })
|
||||
.where(eq(nativeRunFinalizations.runId, run.id));
|
||||
await tx.update(environmentLeases).set({ metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || ${JSON.stringify({
|
||||
runnerTaskRetiring: true, runnerTaskStartupCleanup: { contextId: marker.contextId, requiresPreparationStop: marker.requiresPreparationStop },
|
||||
})}::jsonb`, updatedAt: new Date() }).where(eq(environmentLeases.id, lease.id));
|
||||
return true;
|
||||
});
|
||||
if (publication && options.publish !== false) publishActivity(publication);
|
||||
return requested;
|
||||
}
|
||||
|
||||
async function assertStartupCreationsStopped(db: Db, marker: NativeStartupCleanup): Promise<void> {
|
||||
const creations = await db.select({ creation: sandboxCreationOperations, lease: environmentLeases }).from(sandboxCreationOperations)
|
||||
.leftJoin(environmentLeases, eq(environmentLeases.id, sandboxCreationOperations.leaseId)).where(and(
|
||||
eq(sandboxCreationOperations.companyId, marker.companyId), eq(sandboxCreationOperations.runId, marker.runId),
|
||||
sql`${sandboxCreationOperations.metadata}->>'runnerTaskLease' = 'true'`));
|
||||
if (creations.some(({ creation, lease }) => creation.state !== "attached" &&
|
||||
(!["cleanup_pending", "resolved"].includes(creation.state) || !lease || lease.companyId !== marker.companyId || lease.heartbeatRunId !== marker.runId ||
|
||||
lease.providerLeaseId !== creation.metadata.creationProviderLeaseId || lease.cleanupStatus !== "success" || !hasRemoteTerminationReceipt(lease)))) {
|
||||
throw conflict("Task cleanup is still pending", { code: NATIVE_STARTUP_CLEANUP_PENDING });
|
||||
}
|
||||
}
|
||||
|
||||
export async function verifyNativeStartupCleanup(db: Db, run: Run, release: () => Promise<unknown>): Promise<void> {
|
||||
const marker = readNativeStartupCleanup(run);
|
||||
if (!marker || isNativeRunnerOwnershipHeld(run) || !["running", "cancelled"].includes(run.status)) throw conflict("Native startup cleanup intent unavailable");
|
||||
const read = async () => {
|
||||
const [lease] = await db.select().from(environmentLeases).where(and(eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.id, marker.leaseId)));
|
||||
if (!lease || lease.heartbeatRunId !== run.id || lease.providerLeaseId !== marker.taskId) throw conflict("Native startup cleanup target changed");
|
||||
return lease;
|
||||
};
|
||||
const [owner] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id));
|
||||
if (!owner || owner.controllerGeneration !== marker.controllerGeneration || owner.attempt !== marker.attempt ||
|
||||
owner.leaseOwner !== marker.leaseOwner || owner.resultId) throw conflict("Native startup cleanup ownership changed");
|
||||
let lease = await read();
|
||||
const stopped = (value: typeof lease) => hasRemoteTerminationReceipt(value) && value.cleanupStatus === "success" &&
|
||||
(!marker.requiresPreparationStop || value.metadata?.runnerTaskPreparationStopped === true);
|
||||
if (!stopped(lease)) { await release(); lease = await read(); }
|
||||
if (!stopped(lease)) throw conflict("Task cleanup is still pending", { code: NATIVE_STARTUP_CLEANUP_PENDING });
|
||||
await assertStartupCreationsStopped(db, marker);
|
||||
await db.transaction(async tx => {
|
||||
await tx.select({ id: issues.id }).from(issues).where(and(eq(issues.companyId, run.companyId), eq(issues.id, run.nativeIssueId!))).for("update");
|
||||
const [coordinator] = await tx.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id)).for("update");
|
||||
const [current] = await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)).for("update");
|
||||
const observed = current && readNativeStartupCleanup(current);
|
||||
const [receipt] = await tx.select().from(environmentLeases).where(eq(environmentLeases.id, marker.leaseId)).for("update");
|
||||
if (!coordinator || !current || isNativeRunnerOwnershipHeld(current) || coordinator.resultId || coordinator.controllerGeneration !== marker.controllerGeneration ||
|
||||
coordinator.attempt !== marker.attempt || coordinator.leaseOwner !== marker.leaseOwner || observed?.contextId !== marker.contextId ||
|
||||
!receipt || receipt.providerLeaseId !== marker.taskId || receipt.heartbeatRunId !== run.id || !stopped(receipt)) {
|
||||
throw conflict("Native startup cleanup ownership changed");
|
||||
}
|
||||
await assertStartupCreationsStopped(tx as unknown as Db, marker);
|
||||
await tx.update(heartbeatRuns).set({ resultJson: sql`jsonb_set(${heartbeatRuns.resultJson}, '{nativeStartupCleanup,state}', '"complete"'::jsonb)`, updatedAt: new Date() })
|
||||
.where(eq(heartbeatRuns.id, run.id));
|
||||
});
|
||||
}
|
||||
|
||||
export async function acknowledgeNativeStartupCleanup(db: Db, input: { companyId: string; runId: string }): Promise<void> {
|
||||
const [binding] = await db.select({ issueId: nativeRunFinalizations.issueId }).from(nativeRunFinalizations).where(and(
|
||||
eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.runId, input.runId)));
|
||||
if (!binding) return;
|
||||
await db.transaction(async tx => {
|
||||
await tx.select({ id: issues.id }).from(issues).where(and(eq(issues.companyId, input.companyId), eq(issues.id, binding.issueId))).for("update");
|
||||
const [owner] = await tx.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, input.runId)).for("update");
|
||||
const [run] = await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId)).for("update");
|
||||
const marker = run && readNativeStartupCleanup(run);
|
||||
if (!owner || !run || run.status !== "cancelled" || !marker || marker.state !== "complete" || owner.resultId ||
|
||||
owner.failureCode !== NATIVE_STARTUP_CLEANUP_PENDING || owner.controllerGeneration !== marker.controllerGeneration ||
|
||||
owner.attempt !== marker.attempt || owner.leaseOwner !== marker.leaseOwner) return;
|
||||
const [lease] = await tx.select().from(environmentLeases).where(eq(environmentLeases.id, marker.leaseId)).for("update");
|
||||
if (!lease || lease.heartbeatRunId !== run.id || lease.providerLeaseId !== marker.taskId || !hasRemoteTerminationReceipt(lease) ||
|
||||
(marker.requiresPreparationStop && lease.metadata?.runnerTaskPreparationStopped !== true)) return;
|
||||
await assertStartupCreationsStopped(tx as unknown as Db, marker);
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle);
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.environmentCleanup);
|
||||
await tx.update(nativeRunFinalizations).set({ leaseOwner: null, leaseExpiresAt: null, controlDeadlineAt: null, recoveryState: null,
|
||||
failureCode: "native_retry_cancelled", updatedAt: new Date() }).where(and(eq(nativeRunFinalizations.runId, input.runId), isNull(nativeRunFinalizations.resultId)));
|
||||
});
|
||||
}
|
||||
Reference in new issue
Block a user