diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index aa0ecf30a0a..9d23e4d847d 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -3422,6 +3422,13 @@ and `packages/adapter-utils/src/sandbox-callback-bridge.test.ts` for these contr ### Native runner restart recovery +An explicit Retry can continue a recovered setup failure that began as a +scheduled retry. Its settlement proof accepts the host-authored promotion event +only when its deadline, reason and attempt match the run and its saved retry +context. The exact failure and durable measurement must still follow. Provider +dispatch evidence, an active controller, unsafe files or changed user authority +keep the Retry held. This adds no automatic replay or recovery loop. + For remote controller disk-loss work, use the existing control-plane snapshot transaction and `native-controller-recovery.ts`; do not add a second snapshot writer. The private versioned metadata is limited to 4 KiB. Its importer handles diff --git a/server/src/services/failed-preparation-settlement.ts b/server/src/services/failed-preparation-settlement.ts index a9d6551a2fc..3fb3844b73a 100644 --- a/server/src/services/failed-preparation-settlement.ts +++ b/server/src/services/failed-preparation-settlement.ts @@ -1,4 +1,4 @@ -import { and, asc, eq } from "drizzle-orm"; +import { and, asc, eq, sql } from "drizzle-orm"; import { heartbeatRunEvents, heartbeatRuns, type Db } from "@paperclipai/db"; import { isUuidLike } from "@paperclipai/shared"; @@ -10,19 +10,25 @@ export async function readFailedPreparationSettlement(db: Db, source: Source, al if (typeof source.error !== "string" || !source.error.length) return null; const events = await db.select().from(heartbeatRunEvents).where(and( eq(heartbeatRunEvents.companyId, source.companyId), eq(heartbeatRunEvents.runId, source.id), - )).orderBy(asc(heartbeatRunEvents.seq)).limit(allowRunStarted ? 5 : 4); + )).orderBy(asc(heartbeatRunEvents.seq)).limit(allowRunStarted ? 6 : 5); const hostEvent = (event: typeof heartbeatRunEvents.$inferSelect) => event.agentId === source.agentId && event.stream === "system" && event.sourceInstanceId === null && event.sourceEventId === null && event.sourceSeq === null && event.sourcePayloadSha256 === null && event.protocolSchemaVersion === null; let offset = 0; + const promotion = events[0]; + if (promotion?.message === "Scheduled retry became due and was promoted to the queued run pool") { + if (!hostEvent(promotion) || promotion.seq !== 1 || promotion.eventType !== "lifecycle" + || promotion.level !== "info" || !await matchesScheduledRetryPromotion(db, source, promotion)) return null; + offset++; + } let unsupportedHarnessWarning: { message: string | null; revision: string } | null = null; - const warning = events[0]; + const warning = events[offset]; if (warning?.eventType === "agent_observability.unavailable") { - if (!hostEvent(warning) || warning.seq !== 1 || warning.level !== "warn" + if (!hostEvent(warning) || warning.seq !== 1 + offset || warning.level !== "warn" || warning.payload?.reason !== "unsupported_harness" || typeof warning.payload.revision !== "string" || !isUuidLike(warning.payload.revision) || Object.keys(warning.payload).some(key => !["reason", "revision"].includes(key))) return null; unsupportedHarnessWarning = { message: warning.message, revision: warning.payload.revision }; - offset = 1; + offset++; } const started = events[offset]; if (allowRunStarted && started?.eventType === "lifecycle" && started.message === "run started") { @@ -41,3 +47,29 @@ export async function readFailedPreparationSettlement(db: Db, source: Source, al || !Number.isFinite(measured.payload.durationMs) || measured.payload.durationMs < 0) return null; return { unsupportedHarnessWarning }; } + +async function matchesScheduledRetryPromotion(db: Db, source: Source, event: typeof heartbeatRunEvents.$inferSelect) { + const [run] = await db.select({ retryOfRunId: heartbeatRuns.retryOfRunId, + scheduledRetryAt: heartbeatRuns.scheduledRetryAt, scheduledRetryAttempt: heartbeatRuns.scheduledRetryAttempt, + scheduledRetryReason: heartbeatRuns.scheduledRetryReason, + retryContext: sql>`jsonb_build_object( + 'retryOfRunId', ${heartbeatRuns.contextSnapshot}->'retryOfRunId', + 'retryReason', ${heartbeatRuns.contextSnapshot}->'retryReason', + 'scheduledRetryAt', ${heartbeatRuns.contextSnapshot}->'scheduledRetryAt', + 'scheduledRetryAttempt', ${heartbeatRuns.contextSnapshot}->'scheduledRetryAttempt')`, + }).from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, source.companyId), + eq(heartbeatRuns.agentId, source.agentId), eq(heartbeatRuns.id, source.id))); + const due = run?.scheduledRetryAt?.toISOString(); + return Boolean(run && run.retryOfRunId && isUuidLike(run.retryOfRunId) && due + && Number.isSafeInteger(run.scheduledRetryAttempt) && run.scheduledRetryAttempt! > 0 + && typeof run.scheduledRetryReason === "string" && run.scheduledRetryReason.length > 0 + && run.retryContext.retryOfRunId === run.retryOfRunId + && run.retryContext.retryReason === run.scheduledRetryReason + && run.retryContext.scheduledRetryAt === due + && run.retryContext.scheduledRetryAttempt === run.scheduledRetryAttempt + && event.createdAt >= run.scheduledRetryAt! + && event.payload?.scheduledRetryAt === due + && event.payload.scheduledRetryAttempt === run.scheduledRetryAttempt + && event.payload.scheduledRetryReason === run.scheduledRetryReason + && Object.keys(event.payload).every(key => ["scheduledRetryAt", "scheduledRetryAttempt", "scheduledRetryReason"].includes(key))); +} diff --git a/server/src/services/workspace-recovery-scheduled-preparation.test.ts b/server/src/services/workspace-recovery-scheduled-preparation.test.ts new file mode 100644 index 00000000000..055e6a96ed5 --- /dev/null +++ b/server/src/services/workspace-recovery-scheduled-preparation.test.ts @@ -0,0 +1,101 @@ +import { randomUUID } from "node:crypto"; +import { readFile } from "node:fs/promises"; +import path from "node:path"; +import { and, eq, sql } from "drizzle-orm"; +import { agentWakeupRequests, heartbeatRunEvents, heartbeatRuns } from "@paperclipai/db"; +import type { AdapterSandboxExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; +import request from "supertest"; +import { describe, expect, it, vi } from "vitest"; +import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js"; +import { heartbeatService } from "./heartbeat.js"; +import { readFreshWorkspaceReadiness } from "./fresh-workspace-continuation.js"; +import { recoverRetainedWorkspace } from "./workspace-recovery.js"; +import { createSavedWorkspaceLoader } from "./remote-workspaces.js"; +import { boardRecoveryApp, db, failedRemotePreparation, receiptOnlyRecoveryApp, run, useWorkspaceRecoveryFixture } from "./__tests__/workspace-recovery-fixture.js"; + +const targets = vi.hoisted(() => new Map()); +vi.mock("./environment-execution-target.js", async importOriginal => { + const actual = await importOriginal(); + return { ...actual, resolveEnvironmentExecutionTarget: async (input: Parameters[0]) => + (input.leaseId ? targets.get(input.leaseId) : undefined) ?? actual.resolveEnvironmentExecutionTarget(input) }; +}); + +describe("recovered scheduled preparation", () => { + useWorkspaceRecoveryFixture(targets); + + async function scheduledPreparation(warning = false) { + const f = await failedRemotePreparation("retained", false, warning), source = await run(f.runId); + const events = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.runId)).orderBy(heartbeatRunEvents.seq); + const [parent] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: source.agentId, + issueId: f.issueId, status: "failed", finishedAt: new Date(source.createdAt.getTime() - 1000), + createdAt: new Date(source.createdAt.getTime() - 2000) }).returning(); + const due = new Date(Date.now() - 1000), reason = "ai_connection_pool_wait"; + await db.update(heartbeatRuns).set({ retryOfRunId: parent.id, scheduledRetryAt: due, + scheduledRetryReason: reason, scheduledRetryAttempt: 1, sessionIdBefore: randomUUID(), + contextSnapshot: { ...source.contextSnapshot, retryOfRunId: parent.id, retryReason: reason, + scheduledRetryAt: due.toISOString(), scheduledRetryAttempt: 1 }, + }).where(eq(heartbeatRuns.id, f.runId)); + await db.delete(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.runId)); + await db.update(heartbeatRuns).set({ nextEventSeq: 1 }).where(eq(heartbeatRuns.id, f.runId)); + await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId: f.runId, agentId: source.agentId, + eventType: "lifecycle", stream: "system", level: "info", + message: "Scheduled retry became due and was promoted to the queued run pool", + payload: { scheduledRetryAt: due.toISOString(), scheduledRetryAttempt: 1, scheduledRetryReason: reason } }); + for (const event of events) await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId: f.runId, + agentId: source.agentId, eventType: event.eventType, stream: event.stream, level: event.level, + message: event.message, payload: event.payload }); + await recoverRetainedWorkspace(f.input); + return f; + } + + it.each([false, true])("delivers one HTTP Retry after a scheduled setup failure has recovered files (warning=%s)", async warning => { + const f = await scheduledPreparation(warning), source = await run(f.runId), heartbeat = heartbeatService(db); + const intent = { observedRunId: f.runId, idempotencyKey: randomUUID() }, app = boardRecoveryApp(); + const receipt = (await request(app).post(`/api/issues/${f.issueId}/retry`).send(intent).expect(202)).body; + await Promise.all([heartbeat.resumeTaskRetry(f.companyId, receipt.requestId), heartbeat.resumeTaskRetry(f.companyId, receipt.requestId)]); + expect((await request(app).post(`/api/issues/${f.issueId}/retry`).send(intent).expect(202)).body.requestId).toBe(receipt.requestId); + const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), + sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${receipt.requestId}`)); + expect(successors).toHaveLength(1); + expect(successors[0]).toMatchObject({ status: "queued", contextSnapshot: { previousRunId: f.runId, + taskRetryRequestId: receipt.requestId, forceFreshSession: true } }); + expect(await readFreshWorkspaceReadiness(db, successors[0])).toMatchObject({ mode: "resume_preserved_workspace" }); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.requestId)))[0]) + .toMatchObject({ status: "coalesced", runId: successors[0].id }); + expect(await run(f.runId)).toEqual(source); + const remote = source.runnerProfileJson!.remoteWorkspace as { snapshotId: string }; + const saved = await createSavedWorkspaceLoader(db)({ companyId: f.companyId, + workspaceId: f.workspaceId, snapshotId: remote.snapshotId }); + try { expect(await readFile(path.join(saved.root, "saved.txt"), "utf8")).toBe("finished work\n"); } + finally { await saved.cleanup(); } + }, 30_000); + + it.each(["attempt", "due", "reason", "context", "parent", "foreign", "provider", "extra", "controller", "stop"] as const)( + "retains the saved Retry when scheduled preparation proof changes: %s", async gate => { + const f = await scheduledPreparation(), source = await run(f.runId); + const receipt = (await request(receiptOnlyRecoveryApp()).post(`/api/issues/${f.issueId}/retry`) + .send({ observedRunId: f.runId, idempotencyKey: randomUUID() }).expect(202)).body; + if (["attempt", "due", "reason", "extra"].includes(gate)) await db.update(heartbeatRunEvents).set({ payload: { + scheduledRetryAt: (source.scheduledRetryAt!).toISOString(), scheduledRetryAttempt: 1, + scheduledRetryReason: source.scheduledRetryReason, + ...(gate === "attempt" ? { scheduledRetryAttempt: 2 } : gate === "due" ? { scheduledRetryAt: new Date(0).toISOString() } + : gate === "reason" ? { scheduledRetryReason: "transient_failure" } : { unknown: true }), + } }).where(and(eq(heartbeatRunEvents.runId, f.runId), eq(heartbeatRunEvents.seq, 1))); + if (gate === "context") await db.update(heartbeatRuns).set({ contextSnapshot: { ...source.contextSnapshot, + scheduledRetryAttempt: 2 } }).where(eq(heartbeatRuns.id, f.runId)); + if (gate === "parent") await db.update(heartbeatRuns).set({ retryOfRunId: null }).where(eq(heartbeatRuns.id, f.runId)); + if (gate === "foreign") await db.update(heartbeatRunEvents).set({ sourceEventId: randomUUID() }) + .where(and(eq(heartbeatRunEvents.runId, f.runId), eq(heartbeatRunEvents.seq, 1))); + if (gate === "provider") await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId: f.runId, + agentId: source.agentId, eventType: "adapter.invoke", stream: "system", message: "provider dispatch" }); + if (gate === "controller") await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: new Date(Date.now() + 60_000) }) + .where(eq(heartbeatRuns.id, f.runId)); + if (gate === "stop") await db.update(heartbeatRuns).set({ resultJson: { ...source.resultJson, + cancellation: { source: "operator", expected: true, initiator: { type: "user", userId: "local-board" }, + reason: "Stop", recordedAt: new Date().toISOString() } } }).where(eq(heartbeatRuns.id, f.runId)); + await heartbeatService(db).resumeTaskRetry(f.companyId, receipt.requestId); + expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), + sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${receipt.requestId}`))).toHaveLength(0); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.requestId)))[0].runId).toBeNull(); + }, 30_000); +});