diff --git a/server/src/__tests__/interrupted-preparation-hold-settle.test.ts b/server/src/__tests__/interrupted-preparation-hold-settle.test.ts new file mode 100644 index 00000000000..062497bad2b --- /dev/null +++ b/server/src/__tests__/interrupted-preparation-hold-settle.test.ts @@ -0,0 +1,183 @@ +import { spawn } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { and, eq } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { + activityLog, + agents, + authUsers, + companyMemberships, + companies, + createDb, + environmentLeases, + environments, + heartbeatRunEvents, + heartbeatRuns, + issueRecoveryActions, + issues, +} from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { currentProcessSpaceHash } from "../services/controller-process-identity.js"; +import { heartbeatService } from "../services/heartbeat.js"; +import { settleHeldInterruptedPreparation } from "../services/interrupted-preparation-hold-settlement.js"; +import { legacyControllerBootId } from "../services/legacy-controller-lease.js"; +import { createOwnershipReportState, runOwnershipReconciliationPass } from "../services/recovery/ownership-invariant-report.js"; + +const support = await getEmbeddedPostgresTestSupport(); +const suite = support.supported ? describe : describe.skip; +const KIND = "legacy_interrupted_preparation_hold"; + +async function exitedPid(): Promise { + const child = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { stdio: "ignore" }); + const pid = child.pid!; + await new Promise(resolve => { child.once("exit", () => resolve()); child.kill("SIGKILL"); }); + return pid; +} + +/** + * 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 its own controller was still live and + * the sandbox lease was still cleaning 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 on the shared ownership path", () => { + let db!: ReturnType; + let temporary: Awaited> | 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" } }); + await db.insert(authUsers).values({ id: "local-board", name: "Local operator", email: "local-board@example.invalid", + createdAt: new Date(), updatedAt: new Date() }).onConflictDoNothing(); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: "local-board", membershipRole: "owner" }); + }, 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(); }); + + /** The draining controller, now exited, in this PID space. */ + async function deadController(overrides: Record = {}) { + const bootId = randomUUID(); + return { bootId, process: { bootId, pid: await exitedPid(), hostHash: currentProcessSpaceHash(), + processStartedAt: new Date(Date.now() - 60_000).toISOString(), ...overrides } as Record | null }; + } + + async function interruptedPreparation(options: { controller?: { bootId: string; process: Record | null }; + leaseReleased?: boolean; holdEvidence?: Record } = {}) { + const issueId = randomUUID(), runId = randomUUID(), leaseId = randomUUID(), providerLeaseId = `sandbox-${leaseId}`; + const finishedAt = new Date(Date.now() - 10_000); + const controller = options.controller ?? await deadController(); + // Terminalization projected the task to blocked and kept a receipt so the + // settle can restore exactly that projection. + await db.insert(issues).values({ id: issueId, companyId, title: "Warm follow-up", status: "blocked", + assigneeAgentId: agentId, statusVersion: 2 }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "automation", + responsibleUserId: "local-board", issueId, + status: "interrupted", errorCode: "server_shutdown_interrupted", runtimeMode: "legacy", executionStage: "preparing", + startedAt: new Date(finishedAt.getTime() - 2_000), finishedAt, nextEventSeq: 1, + controllerBootId: controller.bootId, controllerProcess: controller.process, + 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() }, + nativeBootstrapFailureBlock: { runId, previousStatus: "in_progress", statusVersion: 2 }, + ...options.holdEvidence } }).returning(); + return { issueId, runId, holdId: hold!.id }; + } + + const hold = async (id: string) => (await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, id)))[0]!; + const pass = async () => (await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: true })) + .kinds.find(entry => entry.kind === KIND); + + it("replays a preparation a restart interrupted once its controller is proven gone", async () => { + // The heartbeat service registers the resume path that schedules the replay. + heartbeatService(db); + const f = await interruptedPreparation(); + expect(await pass()).toMatchObject({ inProgress: 1, settled: 1 }); + // The task is unblocked and an uncharged interrupted-preparation successor exists. + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toMatchObject({ status: "in_progress" }); + const successors = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, f.runId)); + expect(successors).toHaveLength(1); + expect(successors[0]).toMatchObject({ scheduledRetryReason: "interrupted_preparation" }); + const settled = await hold(f.holdId); + expect(settled).toMatchObject({ status: "resolved", outcome: "false_positive" }); + expect(settled.evidence).toMatchObject({ automaticRecovery: { replay: "allowed" }, + bootstrapPreparationContinuation: "delivered", bootstrapPreparationSettledAt: expect.any(String) }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]!.resultJson) + .toMatchObject({ executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }); + const [audit] = await db.select().from(activityLog).where(and(eq(activityLog.runId, f.runId), + eq(activityLog.action, "issue.recovery_action_resolved"))); + expect(audit?.details).toMatchObject({ reason: "native_bootstrap_stopped_before_provider", ownerProof: "pid_dead" }); + }); + + it.each([ + ["provider work may have started", async () => { + const f = await interruptedPreparation(); + await db.insert(heartbeatRunEvents).values({ companyId, runId: f.runId, agentId, seq: 1, eventType: "turn.started", + stream: "system", level: "info", message: "turn started" }); + return f; + }], + ["its controller stamp is from another process space", async () => + interruptedPreparation({ controller: await deadController({ hostHash: "other-container" }) })], + ["it has no controller stamp", async () => + interruptedPreparation({ controller: { bootId: randomUUID(), process: null } })], + ["its controller is this process", async () => interruptedPreparation({ controller: { bootId: legacyControllerBootId, + process: { bootId: legacyControllerBootId, pid: process.pid, hostHash: currentProcessSpaceHash(), processStartedAt: null } } })], + ["its sandbox lease is still cleaning up", async () => interruptedPreparation({ leaseReleased: false })], + ["an operator confirmed the owner gone on the hold", async () => + interruptedPreparation({ holdEvidence: { ownerGoneAttestation: { actorId: "board-user" } } })], + ["the user already continued", async () => + interruptedPreparation({ holdEvidence: { explicitUserContinuation: { actorId: "board-user" } } })], + ] as const)("keeps the hold when %s", async (_label, arrange) => { + const f = await arrange(); + const before = await hold(f.holdId); + expect((await pass())?.settled ?? 0).toBe(0); + expect(await hold(f.holdId)).toEqual(before); + }); + + it("settles exactly once when two processes judge the same hold", async () => { + const f = await interruptedPreparation(); + const record = { recordId: f.runId, companyId, runId: f.runId, issueId: f.issueId, owner: { kind: "none" as const }, + updatedAt: null }; + const context = { now: new Date(), proof: { verdict: "gone" as const, reason: "pid_dead" as const } }; + const outcomes = await Promise.all([settleHeldInterruptedPreparation(db, record, context), + settleHeldInterruptedPreparation(db, record, context)]); + expect(outcomes.filter(outcome => outcome.settled)).toHaveLength(1); + const resolutions = await db.select().from(activityLog).where(and(eq(activityLog.runId, f.runId), + eq(activityLog.action, "issue.recovery_action_resolved"))); + expect(resolutions).toHaveLength(1); + }); +}); diff --git a/server/src/__tests__/ownership-reconciliation-report.test.ts b/server/src/__tests__/ownership-reconciliation-report.test.ts index f50c4d9dd04..309f0ab683f 100644 --- a/server/src/__tests__/ownership-reconciliation-report.test.ts +++ b/server/src/__tests__/ownership-reconciliation-report.test.ts @@ -144,7 +144,7 @@ describe("report-only ownership reconciliation", () => { expect(new Set(names).size).toBe(names.length); expect(names.length).toBeGreaterThanOrEqual(40); expect(OWNABLE_KINDS.filter(kind => kind.settle !== null).map(kind => kind.id)) - .toEqual(["native_accepted_result_lease", "native_finalizer_lease", "native_cleanup_lease", "sandbox_creation_intent", "environment_run_lease_active", "environment_teardown_claim", "environment_retained_orphan_sandbox", "environment_login_probe_lease", "workspace_runtime_control_operation", "native_workspace_finalize_attempt", "execution_workspace_runtime_lease", "legacy_shutdown_settlement_fence", "copyback_owner_hold", "observed_workspace_preservation_fence", "wake_request_claimed", "legacy_controller_lease", "issue_execution_lock", "issue_checkout_lock", "agent_lifecycle_lease"]); + .toEqual(["native_accepted_result_lease", "native_finalizer_lease", "native_cleanup_lease", "sandbox_creation_intent", "environment_run_lease_active", "environment_teardown_claim", "environment_retained_orphan_sandbox", "environment_login_probe_lease", "workspace_runtime_control_operation", "native_workspace_finalize_attempt", "execution_workspace_runtime_lease", "legacy_shutdown_settlement_fence", "copyback_owner_hold", "observed_workspace_preservation_fence", "wake_request_claimed", "legacy_controller_lease", "legacy_interrupted_preparation_hold", "issue_execution_lock", "issue_checkout_lock", "agent_lifecycle_lease"]); expect(OWNABLE_KINDS.filter(kind => !kind.listInProgress).every(kind => kind.coverageGap)).toBe(true); }); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 19901756c0e..2997874085a 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,4 @@ +import { registerHeldPreparationResumer } from "./interrupted-preparation-hold-settlement.js"; import { resolveProviderApiPlacement } from "./native-runtime/provider-api-placement.js"; import type { StopInvocationOptions } from "./heartbeat/run-control.js"; import { isContextCompactionRun } from "./context-compaction.js"; @@ -1530,6 +1531,12 @@ export function heartbeatService( function resumeRemoteStopComments(...args: Parameters) { return runControl.resumeRemoteStopComments(...args); } + // The interrupted-preparation ownership kind settles a hold, then replays the + // preparation through this same resume path (bounded, uncharged retry). + registerHeldPreparationResumer(async ({ companyId, runId }) => { + const run = await getRun(runId); + if (run && run.companyId === companyId) await resumeRemoteStopComments(run); + }); function cancelRunInternal(...args: Parameters) { return runControl.cancelRunInternal(...args); diff --git a/server/src/services/interrupted-preparation-hold-settlement.ts b/server/src/services/interrupted-preparation-hold-settlement.ts new file mode 100644 index 00000000000..a2ad306f4ef --- /dev/null +++ b/server/src/services/interrupted-preparation-hold-settlement.ts @@ -0,0 +1,84 @@ +import { and, eq, inArray, isNull, sql } from "drizzle-orm"; +import { heartbeatRuns, type Db } from "@paperclipai/db"; +import { isInterruptedNativePreparation } from "./cancelled-native-startup.js"; +import { LEGACY_RECOVERY_CAUSE, settleInterruptedNativeBootstrap } from "./legacy-execution-recovery.js"; +import { logger } from "../middleware/logger.js"; +import type { OwnerProof } from "./owner-proof.js"; +import type { InProgressRecord, SettleOutcome } from "./recovery/ownership-registry.js"; +import { OWNER_GONE_ATTESTATION_KEY } from "./recovery/owner-gone-attestation.js"; + +/** + * `legacy_interrupted_preparation_hold`: a run a graceful shutdown interrupted + * while it was still preparing, held with `preserve_without_replay_v1`. + * + * The draining process terminalizes the run before the no-provider-work proof + * can hold: its own controller is still live and its sandbox lease may still + * be cleaning up. Once the controller is proven gone and every lease is + * released with a confirmed stop, the existing proof decides, and a run with + * no provider execution is settled and replayed in its own uncharged lane. + * + * Only holds the system recorded itself are listed: an operator decision (an + * owner-gone attestation, an explicit user continuation, an assigned owner) is + * never reopened. A run with a lease still cleaning up is not listed yet; the + * next pass lists it once the lease settles. + */ +export function heldInterruptedPreparationWhere() { + return 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 hold.status = 'resolved' and hold.outcome = 'blocked' and hold.owner_user_id is null + and hold.evidence->'automaticRecovery'->>'policy' = 'preserve_without_replay_v1' + and hold.evidence->'automaticRecovery'->>'replay' = 'blocked' + and not (hold.evidence ?| array['executionReconciliation', ${OWNER_GONE_ATTESTATION_KEY}, 'explicitUserContinuation']))`, + sql`not exists (select 1 from environment_leases lease where lease.company_id = ${heartbeatRuns.companyId} + and lease.heartbeat_run_id = ${heartbeatRuns.id} + and (lease.released_at is null or lease.status = 'pending_cleanup' or lease.cleanup_status = 'failed'))`); +} + +/** Continues a settled preparation through the ordinary resume path + * (`resumeRemoteStopComments`): it schedules the uncharged + * `interrupted_preparation` successor and records the continuation. The + * heartbeat service registers it; it owns the scheduling dependencies. */ +type HeldPreparationResumer = (input: { companyId: string; runId: string }) => Promise; +let heldPreparationResumer: HeldPreparationResumer | null = null; + +export function registerHeldPreparationResumer(resumer: HeldPreparationResumer): void { + heldPreparationResumer = resumer; +} + +/** + * Settle of the kind. The shared reconciler calls it only after + * `proveOwnerGone` proved the run's recorded controller gone; a run without a + * controller stamp, or one from another process space, is never settled. It + * re-checks the existing durable proof (no coordinator, result or execution + * events; leases released with stop receipts; workspace repair when needed) + * and resolves the hold through `settleInterruptedNativeBootstrap`, whose row + * locks make the transition, and the continuation it queues, happen once. + */ +export async function settleHeldInterruptedPreparation(db: Db, record: InProgressRecord, + context: { now: Date; proof: OwnerProof }): Promise { + if (!record.companyId) return { settled: false, reason: "run_has_no_company" }; + const [run] = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, record.recordId), + eq(heartbeatRuns.companyId, record.companyId), heldInterruptedPreparationWhere())); + if (!run) return { settled: false, reason: "no_longer_a_held_preparation" }; + if (!await isInterruptedNativePreparation(db, run)) { + return { settled: false, reason: "provider_work_not_disproved", watch: false }; + } + let resolved = 0; + await settleInterruptedNativeBootstrap(db, { run, providerDispatchStarted: false, proof: context.proof.reason, + onHoldsResolved: count => { resolved = count; } }); + if (resolved === 0) return { settled: false, reason: "hold_changed_since_proof" }; + // Only the call that resolved the hold starts the replay, so it starts once. + // Without a registered resumer (no heartbeat service in this process), the + // pending continuation stays recorded for the next resume of this run. + if (heldPreparationResumer) { + await heldPreparationResumer({ companyId: record.companyId, runId: run.id }).catch(error => { + // The hold is settled either way; its pending continuation stays recorded + // for the next resume of this run (a wake, a cleanup completion). + logger.warn({ err: error, runId: run.id }, "settled interrupted preparation could not schedule its replay"); + }); + } + return { settled: true, reason: context.proof.reason }; +} diff --git a/server/src/services/legacy-execution-recovery.ts b/server/src/services/legacy-execution-recovery.ts index 978528a0246..b202cafa4ec 100644 --- a/server/src/services/legacy-execution-recovery.ts +++ b/server/src/services/legacy-execution-recovery.ts @@ -328,7 +328,11 @@ export async function terminalizeLegacyExecution(input: { } export async function settleInterruptedNativeBootstrap( - db: Db, input: { run: Run; providerDispatchStarted: boolean }, + db: Db, input: { run: Run; providerDispatchStarted: boolean; + /** Owner proof that authorized this settle, recorded in the activity log. */ + proof?: string; + /** Called inside the transaction with the number of holds this call resolved. */ + onHoldsResolved?: (count: number) => void }, ): Promise { if (input.providerDispatchStarted) return null; const issueId = input.run.contextSnapshot?.issueId; @@ -386,8 +390,10 @@ export async function settleInterruptedNativeBootstrap( }).where(inArray(issueRecoveryActions.id, holds.map(hold => hold.id))).returning({ id: issueRecoveryActions.id }) : []; if (resolved.length) await logActivity(tx as unknown as Db, { companyId: run.companyId, actorType: "system", actorId: "system", action: "issue.recovery_action_resolved", entityType: "issue", entityId: task.id, agentId: run.agentId, runId: run.id, - details: { reason: "native_bootstrap_stopped_before_provider", recoveryActionIds: resolved.map(r => r.id) }, + details: { reason: "native_bootstrap_stopped_before_provider", recoveryActionIds: resolved.map(r => r.id), + ...(input.proof ? { ownerProof: input.proof } : {}) }, }); + input.onHoldsResolved?.(resolved.length); return settled; }); } diff --git a/server/src/services/recovery/ownership-registry.ts b/server/src/services/recovery/ownership-registry.ts index 92c31db0873..302ba1147cf 100644 --- a/server/src/services/recovery/ownership-registry.ts +++ b/server/src/services/recovery/ownership-registry.ts @@ -46,6 +46,7 @@ import { nativeControllerOwnerStamp, runProcessOwnerStamp } from "../native-runt import { settleIssueLock } from "./issue-lock-settlement.js"; import { settleAgentLifecycleLease } from "./agent-lifecycle-lease-settlement.js"; import { legacyControllerOwnerStamp, settleLegacyControllerLease } from "../legacy-controller-lease.js"; +import { heldInterruptedPreparationWhere, settleHeldInterruptedPreparation } from "../interrupted-preparation-hold-settlement.js"; import { nativeLeaseHolderStamp, readNativeLeaseHolder, @@ -971,6 +972,21 @@ export const OWNABLE_KINDS: readonly OwnableKind[] = [ }, settle: settleLegacyControllerLease, }, + { + // A preparation a shutdown interrupted, held before its no-provider-work + // proof could exist. Its owner is the recorded controller process: the + // shared prover must prove it gone (a missing stamp, or one from another + // host or PID space, is unprovable and stays held). The settle re-checks + // the durable proof and replays the preparation exactly once. + // Keep this entry after the environment teardown kinds: a lease they settle + // in a pass is released before this kind lists in that same pass. + ...runKind("legacy_interrupted_preparation_hold", "Interrupted preparation held for replay", "ownership", + heldInterruptedPreparationWhere(), row => { + const stamp = legacyControllerOwnerStamp(row); + return stamp ? { kind: "process", stamp } : { kind: "none" }; + }), + settle: settleHeldInterruptedPreparation, + }, runKind("detached_process_marker", "Detached-process marker", "ownership", and(eq(heartbeatRuns.status, "running"), eq(heartbeatRuns.errorCode, "process_detached")), row => row.processPid ? { kind: "process", stamp: runProcessStamp(row, "detached_process_marker") } : { kind: "none" },