mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
fix(recovery): replay a preparation a restart interrupted once proof shows no provider work started
A graceful shutdown terminalizes a run still in preparation inside the draining process, before the no-provider-work proof can hold (its own controller is live and its sandbox lease may still be cleaning up), so the run is held with preserve_without_replay and the task waits for a manual Retry. Orphan recovery now re-judges those holds after every restart: with the existing proof (isInterruptedNativePreparation) the hold is settled through settleInterruptedNativeBootstrap and the preparation is replayed in its own uncharged lane. Possible provider work, or missing proof, stays held. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
3448c7f9ba
commit
3e14a2a943
3 files changed
+169
-1
No files matched your search
@@ -0,0 +1,133 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
|
||||
import {
|
||||
activityLog,
|
||||
agents,
|
||||
companies,
|
||||
createDb,
|
||||
environmentLeases,
|
||||
environments,
|
||||
heartbeatRunEvents,
|
||||
heartbeatRuns,
|
||||
issueRecoveryActions,
|
||||
issues,
|
||||
} from "@paperclipai/db";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import { legacyControllerBootId } from "../services/legacy-controller-lease.js";
|
||||
import { heartbeatService } from "../services/heartbeat.js";
|
||||
import { settleHeldInterruptedPreparations } from "../services/legacy-execution-recovery.js";
|
||||
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
const suite = support.supported ? describe : describe.skip;
|
||||
|
||||
/**
|
||||
* The K2-during-preparation shape from Daytona warm wave w-20261108T142906-s9501
|
||||
* (t05, run c99b3a2e): a scheduled retry was 2 s into preparation when a
|
||||
* graceful shutdown interrupted it. The draining process recorded a
|
||||
* preserve-without-replay hold, because at that moment its own controller was
|
||||
* still live and the sandbox lease was still being cleaned up. The lease was
|
||||
* released with a confirmed stop moments later, and the next process booted,
|
||||
* but nothing re-judged the hold, so the task waited for a manual Retry.
|
||||
*/
|
||||
suite("held interrupted preparations after a restart", () => {
|
||||
let db!: ReturnType<typeof createDb>;
|
||||
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
|
||||
let companyId = "", agentId = "", environmentId = "";
|
||||
|
||||
beforeAll(async () => {
|
||||
temporary = await startEmbeddedPostgresTestDatabase("paperclip-interrupted-preparation-");
|
||||
db = createDb(temporary.connectionString);
|
||||
companyId = randomUUID(); agentId = randomUUID(); environmentId = randomUUID();
|
||||
await db.insert(companies).values({ id: companyId, name: "Paperclip", issuePrefix: "IPH",
|
||||
requireBoardApprovalForNewAgents: false });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Coder", role: "engineer", status: "idle",
|
||||
adapterType: "paperclip_runner", adapterConfig: {}, runtimeConfig: {}, permissions: {} });
|
||||
await db.insert(environments).values({ id: environmentId, name: "Sandbox", driver: "sandbox", status: "active",
|
||||
config: { provider: "fake" } });
|
||||
}, 30_000);
|
||||
|
||||
afterEach(async () => {
|
||||
await db.delete(activityLog);
|
||||
await db.delete(issueRecoveryActions);
|
||||
await db.delete(environmentLeases);
|
||||
await db.delete(heartbeatRunEvents);
|
||||
await db.delete(heartbeatRuns);
|
||||
await db.delete(issues);
|
||||
});
|
||||
|
||||
afterAll(async () => { await temporary?.cleanup(); });
|
||||
|
||||
async function interruptedPreparation(options: { controllerBootId?: string; leaseReleased?: boolean } = {}) {
|
||||
const issueId = randomUUID(), runId = randomUUID(), leaseId = randomUUID(), providerLeaseId = `sandbox-${leaseId}`;
|
||||
const finishedAt = new Date(Date.now() - 10_000);
|
||||
await db.insert(issues).values({ id: issueId, companyId, title: "Warm follow-up", status: "in_progress",
|
||||
assigneeAgentId: agentId });
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "automation",
|
||||
status: "interrupted", errorCode: "server_shutdown_interrupted", runtimeMode: "legacy", executionStage: "preparing",
|
||||
startedAt: new Date(finishedAt.getTime() - 2_000), finishedAt, nextEventSeq: 1,
|
||||
controllerBootId: options.controllerBootId ?? randomUUID(),
|
||||
scheduledRetryAttempt: 1, scheduledRetryReason: "transient_failure",
|
||||
contextSnapshot: { issueId },
|
||||
runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } },
|
||||
resultJson: { stopReason: "interrupted", cancellation: { reason: "Server shutdown interrupted execution",
|
||||
source: "shutdown", expected: true, initiator: { type: "system" }, recordedAt: finishedAt.toISOString() } } });
|
||||
const released = options.leaseReleased !== false;
|
||||
await db.insert(environmentLeases).values({ id: leaseId, companyId, environmentId, issueId, heartbeatRunId: runId,
|
||||
status: released ? "released" : "pending_cleanup", cleanupStatus: released ? "success" : "failed",
|
||||
leasePolicy: "reuse_by_environment", provider: "fake", providerLeaseId,
|
||||
releasedAt: released ? new Date(finishedAt.getTime() + 2_000) : null,
|
||||
metadata: released ? { remoteExecutionTermination: { schema: "paperclip.remote-termination.v1", companyId, runId,
|
||||
leaseId, provider: "fake", providerLeaseId, state: "stopped", confirmedAt: finishedAt.toISOString() } } : {} });
|
||||
const [hold] = await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issueId,
|
||||
kind: "execution_reconciliation", cause: "legacy_execution_requires_reconciliation",
|
||||
fingerprint: `legacy-execution:${runId}`, status: "resolved", outcome: "blocked", ownerType: "board",
|
||||
nextAction: "Inspect the stopped provider and recorded actions.", resolvedAt: finishedAt,
|
||||
evidence: { runId, automaticRecovery: { policy: "preserve_without_replay_v1", runId, replay: "blocked",
|
||||
actionOutcome: "unknown", recordedAt: finishedAt.toISOString() } } }).returning();
|
||||
return { issueId, runId, holdId: hold!.id };
|
||||
}
|
||||
|
||||
const hold = async (id: string) => (await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, id)))[0]!;
|
||||
const runRow = async (id: string) => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0]!;
|
||||
|
||||
it("replays a preparation interrupted by a restart once durable proof shows no provider work started", async () => {
|
||||
const f = await interruptedPreparation();
|
||||
await settleHeldInterruptedPreparations(db);
|
||||
const settled = await hold(f.holdId);
|
||||
expect(settled).toMatchObject({ status: "resolved", outcome: "false_positive" });
|
||||
expect(settled.evidence).toMatchObject({ automaticRecovery: { replay: "allowed" },
|
||||
bootstrapPreparationContinuation: "pending", bootstrapPreparationSettledAt: expect.any(String) });
|
||||
expect((await runRow(f.runId)).resultJson).toMatchObject({
|
||||
executionRecovery: { kind: "bootstrap", providerWorkStarted: false } });
|
||||
});
|
||||
|
||||
it("orphan recovery after a restart settles the hold without a user Retry", async () => {
|
||||
const f = await interruptedPreparation();
|
||||
await heartbeatService(db).reapOrphanedRuns();
|
||||
expect(await hold(f.holdId)).toMatchObject({ status: "resolved", outcome: "false_positive" });
|
||||
});
|
||||
|
||||
it.each([
|
||||
["provider work may have started", async (f: { runId: string }) => {
|
||||
await db.insert(heartbeatRunEvents).values({ companyId, runId: f.runId, agentId, seq: 1, eventType: "turn.started",
|
||||
stream: "system", level: "info", message: "turn started" });
|
||||
}],
|
||||
["the controller that owned it is still this process", async (f: { runId: string }) => {
|
||||
await db.update(heartbeatRuns).set({ controllerBootId: legacyControllerBootId }).where(eq(heartbeatRuns.id, f.runId));
|
||||
}],
|
||||
] as const)("keeps the hold when %s", async (_label, arrange) => {
|
||||
const f = await interruptedPreparation();
|
||||
await arrange(f);
|
||||
const before = await hold(f.holdId);
|
||||
await settleHeldInterruptedPreparations(db);
|
||||
expect(await hold(f.holdId)).toEqual(before);
|
||||
});
|
||||
|
||||
it("keeps the hold while the sandbox lease is still being cleaned up", async () => {
|
||||
const f = await interruptedPreparation({ leaseReleased: false });
|
||||
const before = await hold(f.holdId);
|
||||
await settleHeldInterruptedPreparations(db);
|
||||
expect(await hold(f.holdId)).toEqual(before);
|
||||
});
|
||||
});
|
||||
@@ -47,7 +47,7 @@ import {
|
||||
import { remoteTerminationReceipt } from "../remote-execution-termination.js";
|
||||
import { readEnvironmentStopFailure, type PluginEnvironmentStopFailure } from "@paperclipai/plugin-sdk";
|
||||
import { runUsedConversationAdapter } from "../conversation-continuation.js";
|
||||
import { legacyExecutionNeedsReconciliationWithEvidence, terminalizeLegacyExecution } from "../legacy-execution-recovery.js";
|
||||
import { legacyExecutionNeedsReconciliationWithEvidence, settleHeldInterruptedPreparations, terminalizeLegacyExecution } from "../legacy-execution-recovery.js";
|
||||
import { reconcileRetainedWorkspaces } from "../workspace-recovery.js";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import os from "node:os";
|
||||
@@ -2864,6 +2864,11 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
|
||||
});
|
||||
if (!opts?.skipRetainedNativeSessionCleanup) scheduleRetainedNativeSessionCleanup();
|
||||
await reconcileRetainedWorkspaces(db, environmentRuntime, resumeRemoteStopComments);
|
||||
// A shutdown can hold a run it interrupted in preparation before the proof
|
||||
// that no provider work started exists; once it does, replay that run.
|
||||
await settleHeldInterruptedPreparations(db).catch(error => {
|
||||
logger.warn({ err: error }, "failed to re-judge held interrupted preparations");
|
||||
});
|
||||
if (!opts?.skipNativeStatusWakeIntents) await dispatchPendingNativeStatusWakeups().catch((error) => {
|
||||
logger.warn(
|
||||
{ err: error },
|
||||
|
||||
@@ -392,6 +392,36 @@ export async function settleInterruptedNativeBootstrap(
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Re-judges preparation holds once their proof can exist. A graceful shutdown
|
||||
* terminalizes a run still in preparation inside the draining process: its own
|
||||
* controller is live and its sandbox lease may still be cleaning up, so the
|
||||
* no-provider-work proof (`isInterruptedNativePreparation`) cannot hold yet and
|
||||
* the run is held for reconciliation. Once the owning process is gone and the
|
||||
* leases are released with confirmed stops, that proof decides: a run with no
|
||||
* provider execution is settled and its preparation is replayed in its own
|
||||
* uncharged lane. Anything with possible provider work, or without proof, stays
|
||||
* held. Recovery runs this after every restart.
|
||||
*/
|
||||
export async function settleHeldInterruptedPreparations(db: Db, options: { limit?: number } = {}): Promise<number> {
|
||||
const candidates = await db.select().from(heartbeatRuns).where(and(
|
||||
inArray(heartbeatRuns.status, ["interrupted", "failed"]), eq(heartbeatRuns.runtimeMode, "legacy"),
|
||||
isNull(heartbeatRuns.runtimeModeResolvedAt), eq(heartbeatRuns.executionStage, "preparing"),
|
||||
sql`exists (select 1 from issue_recovery_actions hold where hold.company_id = ${heartbeatRuns.companyId}
|
||||
and hold.cause = ${LEGACY_RECOVERY_CAUSE} and hold.fingerprint = 'legacy-execution:' || ${heartbeatRuns.id}::text
|
||||
and hold.evidence->>'runId' = ${heartbeatRuns.id}::text and not (hold.evidence ? 'executionReconciliation')
|
||||
and (hold.status in ('active', 'escalated') or (hold.status = 'resolved' and hold.outcome = 'blocked'
|
||||
and hold.evidence->'automaticRecovery'->>'policy' = 'preserve_without_replay_v1'
|
||||
and hold.evidence->'automaticRecovery'->>'replay' = 'blocked')))`,
|
||||
)).orderBy(heartbeatRuns.finishedAt).limit(options.limit ?? 50);
|
||||
let settled = 0;
|
||||
for (const run of candidates) {
|
||||
if (!await isInterruptedNativePreparation(db, run)) continue;
|
||||
if (await settleInterruptedNativeBootstrap(db, { run, providerDispatchStarted: false })) settled += 1;
|
||||
}
|
||||
return settled;
|
||||
}
|
||||
|
||||
/** The source stays discoverable until normal admission records a successor.
|
||||
* Known admission gates wait for eligibility events; unresolved writes retain
|
||||
* the pending delivery deadline. Concurrent successful delivery never regresses. */
|
||||
|
||||
Reference in new issue
Block a user