diff --git a/server/src/__tests__/interrupted-preparation-hold-settle.test.ts b/server/src/__tests__/interrupted-preparation-hold-settle.test.ts index 44bd3497b36..062497bad2b 100644 --- a/server/src/__tests__/interrupted-preparation-hold-settle.test.ts +++ b/server/src/__tests__/interrupted-preparation-hold-settle.test.ts @@ -5,6 +5,8 @@ import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; import { activityLog, agents, + authUsers, + companyMemberships, companies, createDb, environmentLeases, @@ -16,6 +18,7 @@ import { } 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"; @@ -55,6 +58,9 @@ suite("held interrupted preparations on the shared ownership path", () => { 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 () => { @@ -80,9 +86,12 @@ suite("held interrupted preparations on the shared ownership path", () => { 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 }); + // 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, @@ -103,7 +112,9 @@ suite("held interrupted preparations on the shared ownership path", () => { 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() }, ...options.holdEvidence } }).returning(); + actionOutcome: "unknown", recordedAt: finishedAt.toISOString() }, + nativeBootstrapFailureBlock: { runId, previousStatus: "in_progress", statusVersion: 2 }, + ...options.holdEvidence } }).returning(); return { issueId, runId, holdId: hold!.id }; } @@ -112,12 +123,19 @@ suite("held interrupted preparations on the shared ownership path", () => { .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: "pending", bootstrapPreparationSettledAt: expect.any(String) }); + 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), 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 index 8eb29278164..a2ad306f4ef 100644 --- a/server/src/services/interrupted-preparation-hold-settlement.ts +++ b/server/src/services/interrupted-preparation-hold-settlement.ts @@ -2,6 +2,7 @@ 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"; @@ -36,6 +37,17 @@ export function heldInterruptedPreparationWhere() { 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 @@ -57,6 +69,16 @@ export async function settleHeldInterruptedPreparation(db: Db, record: InProgres 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" }; + 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 }; }