diff --git a/server/src/__tests__/interrupted-preparation-hold-settle.test.ts b/server/src/__tests__/interrupted-preparation-hold-settle.test.ts index 45062ed7114..44bd3497b36 100644 --- a/server/src/__tests__/interrupted-preparation-hold-settle.test.ts +++ b/server/src/__tests__/interrupted-preparation-hold-settle.test.ts @@ -1,5 +1,6 @@ +import { spawn } from "node:child_process"; import { randomUUID } from "node:crypto"; -import { eq } from "drizzle-orm"; +import { and, eq } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; import { activityLog, @@ -14,23 +15,32 @@ import { issues, } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { currentProcessSpaceHash } from "../services/controller-process-identity.js"; +import { settleHeldInterruptedPreparation } from "../services/interrupted-preparation-hold-settlement.js"; import { legacyControllerBootId } from "../services/legacy-controller-lease.js"; -import { heartbeatService } from "../services/heartbeat.js"; -import { settleHeldInterruptedPreparations } from "../services/legacy-execution-recovery.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 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. + * 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 after a restart", () => { +suite("held interrupted preparations on the shared ownership path", () => { let db!: ReturnType; let temporary: Awaited> | null = null; let companyId = "", agentId = "", environmentId = ""; @@ -58,15 +68,24 @@ suite("held interrupted preparations after a restart", () => { afterAll(async () => { await temporary?.cleanup(); }); - async function interruptedPreparation(options: { controllerBootId?: string; leaseReleased?: boolean } = {}) { + /** 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(); 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(), + controllerBootId: controller.bootId, controllerProcess: controller.process, scheduledRetryAttempt: 1, scheduledRetryReason: "transient_failure", contextSnapshot: { issueId }, runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } }, @@ -84,50 +103,63 @@ suite("held interrupted preparations after a restart", () => { 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(); + actionOutcome: "unknown", recordedAt: finishedAt.toISOString() }, ...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 runRow = async (id: string) => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0]!; + const pass = async () => (await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: true })) + .kinds.find(entry => entry.kind === KIND); - it("replays a preparation interrupted by a restart once durable proof shows no provider work started", async () => { + it("replays a preparation a restart interrupted once its controller is proven gone", async () => { const f = await interruptedPreparation(); - await settleHeldInterruptedPreparations(db); + expect(await pass()).toMatchObject({ inProgress: 1, settled: 1 }); 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" }); + 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 (f: { runId: string }) => { + ["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; }], - ["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)); - }], + ["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 interruptedPreparation(); - await arrange(f); + const f = await arrange(); const before = await hold(f.holdId); - await settleHeldInterruptedPreparations(db); + expect((await pass())?.settled ?? 0).toBe(0); 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); + 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 68c853cac8c..3ff96417cff 100644 --- a/server/src/__tests__/ownership-reconciliation-report.test.ts +++ b/server/src/__tests__/ownership-reconciliation-report.test.ts @@ -143,7 +143,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/recovery.ts b/server/src/services/heartbeat/recovery.ts index 98cb262a72e..71a86bd8c7a 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -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, settleHeldInterruptedPreparations, terminalizeLegacyExecution } from "../legacy-execution-recovery.js"; +import { legacyExecutionNeedsReconciliationWithEvidence, terminalizeLegacyExecution } from "../legacy-execution-recovery.js"; import { reconcileRetainedWorkspaces } from "../workspace-recovery.js"; import { randomUUID } from "node:crypto"; import os from "node:os"; @@ -2864,11 +2864,6 @@ 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 }, 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..8eb29278164 --- /dev/null +++ b/server/src/services/interrupted-preparation-hold-settlement.ts @@ -0,0 +1,62 @@ +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 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'))`); +} + +/** + * 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; } }); + return resolved > 0 ? { settled: true, reason: context.proof.reason } + : { settled: false, reason: "hold_changed_since_proof" }; +} diff --git a/server/src/services/legacy-execution-recovery.ts b/server/src/services/legacy-execution-recovery.ts index 15a6ed25d66..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,42 +390,14 @@ 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; }); } -/** - * 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 { - 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. */ diff --git a/server/src/services/recovery/ownership-registry.ts b/server/src/services/recovery/ownership-registry.ts index 92c31db0873..01e164776db 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,19 @@ 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. + ...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" },