diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index e0e66ab3a32..a2eebcada7c 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -3633,6 +3633,34 @@ re-read dependency readiness, since new dependencies need not change the displayed task status. Old blocked rows without a receipt remain held; no historical status backfill is performed. +Older native executors recorded the local callback-listener startup guard as +`native_session_interrupted`. A saved exact guard outcome can enter the same +bounded safe-replacement path after restart. This requires the first attempt, +an unused provider-session identity, matching saved execution bindings, no +provider events, checkpoint, admission receipt, tool call, or cancellation, and +the existing current snapshot and confirmed stop receipts. A generic error or +a pending provider-pack receipt alone grants no recovery authority. The same +transaction can retire that run's completed file-preservation hold only when +its bound recovery and finalization receipts prove completion. Unrelated holds, +later task blocks, normal dispatch admission, and the three-attempt budget remain +effective. + +Historical replacement discovery reads at most 100 candidates per pass and +keeps a cursor only for the remaining scan. A work notification preserves forward +progress and requests a final rescan, so held earlier rows cannot hide a later +eligible run. Completed scans of held history become quiet. Final snapshot +publication, verified lease cleanup, local process-stop proof, completed workspace +recovery, and dispatch eligibility changes notify the existing replacement worker +in their transactions. A cleared historical interaction only requests another +proof check; it grants no new continuation authority. Source provider effects, +task pause or cancellation, and the original authority checks still prevent replay. +Temporary decision failures retain the existing worker's retry and advance the +bounded cursor, with a forced rescan of earlier failures. A failed page cannot +hide later ready runs. During drain, the initialized replacement worker can read +readiness once per notification generation to discharge an empty notification +without starting replacement. Concrete pending work and changes arriving during +that read remain outstanding. Restart reconstructs the scan from durable records. + For legacy GitHub input without a saved App ID or its own response receipt, Task Retry may prove the endpoint's earlier non-null App claim from a processed response ledger created before that input. The proof stays within the same diff --git a/server/src/__tests__/execution-control-work.test.ts b/server/src/__tests__/execution-control-work.test.ts index 584eeafb6af..345bf605776 100644 --- a/server/src/__tests__/execution-control-work.test.ts +++ b/server/src/__tests__/execution-control-work.test.ts @@ -61,6 +61,41 @@ describe("execution-control lifecycle", () => { expect(run).toHaveBeenCalledTimes(runs + 1); expect(c.nextWakeAt()).toBeNull(); }); + it.each(["empty", "pending", "concurrent"] as const)("settles only proved-empty replacement notifications during drain (%s)", async kind => { + vi.useFakeTimers(); const owner = fakeOwner(); let draining = false; + const c = coordinator(owner, () => !draining); + const run = vi.fn(async () => {}), next = vi.fn(async (): Promise => null); + await registerExecutionControlWork(c, [{ queue: DELIVERY_QUEUES.executionReplacement, run, next, + settleEmptyWhilePaused: true }]).ready; + const before = idleWorkSnapshot().active; + draining = true; + let release: (() => void) | undefined; + if (kind === "pending") next.mockResolvedValue(Date.now() + 15_000); + if (kind === "concurrent") next.mockImplementationOnce(() => new Promise(resolve => { + release = () => resolve(null); + })); + await owner.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.executionReplacement)); + await vi.advanceTimersByTimeAsync(15_000); + if (kind === "concurrent") { + expect(release).toBeTypeOf("function"); + next.mockResolvedValue(Date.now() + 15_000); + await owner.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.executionReplacement)); + release!(); await vi.advanceTimersByTimeAsync(1); + } + expect(run).toHaveBeenCalledTimes(1); + expect(idleWorkSnapshot().active).toBe(before + (kind === "empty" ? 0 : 1)); + const reads = next.mock.calls.length; + await vi.advanceTimersByTimeAsync(60_000); + expect(next).toHaveBeenCalledTimes(reads); + expect(run).toHaveBeenCalledTimes(1); + if (kind !== "empty") { + draining = false; next.mockResolvedValue(null); + await vi.advanceTimersByTimeAsync(15_000); + expect(run).toHaveBeenCalledTimes(2); + } + expect(idleWorkSnapshot().active).toBe(before); + expect(c.nextWakeAt()).toBeNull(); + }); it("keeps a change arriving during a sweep and does not overlap it", async () => { vi.useFakeTimers(); const owner = fakeOwner(); const c = coordinator(owner); let release!: () => void; diff --git a/server/src/__tests__/helpers/historical-listener-continuation.ts b/server/src/__tests__/helpers/historical-listener-continuation.ts new file mode 100644 index 00000000000..332003b42ee --- /dev/null +++ b/server/src/__tests__/helpers/historical-listener-continuation.ts @@ -0,0 +1,134 @@ +import { createServer } from "node:http"; +import { spawn } from "node:child_process"; +import { createRequire } from "node:module"; +import { join, resolve } from "node:path"; +import { mkdir, readFile, writeFile } from "node:fs/promises"; +import { eq } from "drizzle-orm"; +import { completionContracts, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; +import { expect } from "vitest"; +import { createRunDispatch } from "../../modules/run-dispatch/index.js"; +import { prepareNativeHeartbeatRun } from "../../services/native-runtime/prepare-native-run.js"; +import { buildNativeExecutionInput } from "../../services/native-runtime/native-execution-input.js"; +import { buildNativeCompletionContract } from "../../services/native-runtime/completion-contracts.js"; +import { nativeRuntimeContextFixture } from "../../services/native-runtime/runtime-context.test-fixture.js"; +import { materializeAsset } from "../../services/native-runtime/runtime-context.js"; +import { executePaperclipNativeSession } from "../../services/native-runtime/native-session-executor.js"; +import { reconcileNativeFinalizations } from "../../services/native-runtime/native-finalization-reconciler.js"; +import { canonicalNativeRuntimeContextDigest, createNativeSessionBackend, createRunnerdCodexTransport, defaultCapabilityRunnerdBinary } from "../../vendor/paperclip-runner/index.js"; +import { registerLocalRuntimeApiServer } from "../../runtime-api.js"; +import { registerRunnerPrpAuthority, runnerPrpWebSocketInternals, setupRunnerPrpWebSocketServer } from "../../realtime/runner-prp-ws.js"; + +async function prepare(db: Db, runId: string, cwd: string, environmentLeaseId: string) { + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + const [issue] = await db.select().from(issues).where(eq(issues.id, run.nativeIssueId ?? String(run.contextSnapshot?.issueId))); + await db.update(issues).set({ executionRunId: run.id }).where(eq(issues.id, issue.id)); + const native = await prepareNativeHeartbeatRun({ db, run, issue, environmentLeaseId }); + const [contract] = await db.select().from(completionContracts).where(eq(completionContracts.issueId, issue.id)); + const runtimeContext = nativeRuntimeContextFixture(); + runtimeContext.instructions.bundle = await materializeAsset([{ path: "AGENTS.md", mode: 0o444, + content: Buffer.from("Complete the saved task once.\n") }]); + runtimeContext.aggregateDigest = canonicalNativeRuntimeContextDigest(runtimeContext); + const execution = buildNativeExecutionInput({ companyId: run.companyId, agentId: run.agentId, runId, issue, + taskPrompt: "Preserve the saved file and complete the task once.", provider: "codex", + workspace: { id: issue.executionWorkspaceId!, cwd, repoUrl: null, repoRef: null, branchName: null }, + normalizedSessionId: native.normalizedSessionId, runtimeContext, + completionContract: { id: contract.id, sha256: contract.canonicalSha256, schemaVersion: "paperclip.completion-contract.v1", + contract: buildNativeCompletionContract(issue, { revision: contract.revision }) }, + }); + const [current] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...run.runnerProfileJson, ...current.runnerProfileJson, nativeExecutionInput: execution } }) + .where(eq(heartbeatRuns.id, runId)); + return { run: current, native, execution }; +} + +export async function recordHistoricalListenerFailure(input: { + db: Db; connectionString: string; runId: string; cwd: string; environmentId: string; leaseId: string; +}) { + await input.db.update(heartbeatRuns).set({ status: "running", finishedAt: null, error: null, errorCode: null }) + .where(eq(heartbeatRuns.id, input.runId)); + const { execution, native } = await prepare(input.db, input.runId, input.cwd, input.leaseId); + await input.db.insert(nativeRunFinalizations).values({ runId: input.runId, companyId: execution.binding.companyId, + issueId: execution.binding.issueId, phase: "observed" }); + const child = spawn(process.execPath, ["--import", createRequire(import.meta.url).resolve("tsx"), + resolve(import.meta.dirname, "native-drain-controller.test-worker.ts")], { stdio: ["ignore", "pipe", "pipe", "ipc"], env: process.env }); + const logs: string[] = []; + child.stdout?.on("data", data => logs.push(String(data))); + child.stderr?.on("data", data => logs.push(String(data))); + const exited = new Promise(resolveExit => child.once("exit", resolveExit)); + try { + const result = await new Promise((resolveResult, reject) => { + const timeout = setTimeout(() => reject(new Error(logs.join("").slice(-3000))), 60_000); + child.once("error", reject); + child.on("message", (message: { ready?: boolean; done?: boolean; error?: string }) => { + if (message.ready) child.send({ connectionString: input.connectionString, options: { execution, + runnerInstanceId: native.runnerInstanceId, useRunnerd: true, + runnerExecutionTarget: { kind: "remote", transport: "sandbox", environmentId: input.environmentId, + leaseId: input.leaseId, remoteCwd: input.cwd } } }); + if (message.done || message.error) { clearTimeout(timeout); message.error ? reject(new Error(message.error)) : resolveResult(message); } + }); + }); + expect(result).toMatchObject({ commands: 0, errorMessage: "local_runtime_api_listener_unavailable" }); + expect(await exited).toBe(0); + const [failed] = await input.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, input.runId)); + expect(failed).toMatchObject({ phase: "terminal_failure", attempt: 1, failureCode: "native_session_interrupted", + failureDetail: { message: "local_runtime_api_listener_unavailable", recoveryMode: "ambiguous_state", + providerSessionEstablished: false, providerEventsExist: false, checkpointExists: false } }); + } finally { child.kill("SIGKILL"); await exited; } +} + +export async function completeHistoricalListenerSuccessor(input: { db: Db; runId: string; root: string; + prepareWorkspace: () => Promise<{ cwd: string; finalize: () => Promise }> }) { + const [scheduled] = await input.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId)); + const dispatch = createRunDispatch(input.db); + expect(await dispatch.promoteScheduledRetry({ companyId: scheduled.companyId, runId: scheduled.id, + now: new Date(scheduled.scheduledRetryAt!.getTime() + 1) })).toMatchObject({ outcome: "promoted" }); + expect(await dispatch.cancelStaleQueuedRun({ companyId: scheduled.companyId, runId: scheduled.id, expectedStatus: "queued" })) + .toMatchObject({ outcome: "not_stale" }); + await input.db.update(heartbeatRuns).set({ status: "running", runtimeMode: "native", + nativeIssueId: String(scheduled.contextSnapshot?.issueId) }).where(eq(heartbeatRuns.id, scheduled.id)); + await input.db.update(issues).set({ executionRunId: scheduled.id }).where(eq(issues.id, String(scheduled.contextSnapshot?.issueId))); + const workspace = await input.prepareWorkspace(); + const { execution, native } = await prepare(input.db, scheduled.id, workspace.cwd, `controlled-${scheduled.id}`); + await input.db.insert(nativeRunFinalizations).values({ runId: scheduled.id, companyId: scheduled.companyId, + issueId: execution.binding.issueId, phase: "observed" }); + const home = join(input.root, "codex-home"), report = join(input.root, "result.json"), calls = join(input.root, "calls.log"); + await mkdir(home, { recursive: true }); + await writeFile(report, JSON.stringify({ schema: "paperclip.run_result.v1", reportedWorkDisposition: "done", + summary: "Saved task completed after historical controller failure.", completionClaim: { + contractRevision: execution.completionContract.contract.revision, objectiveSatisfied: true, + criteria: execution.completionContract.contract.criteria.map(criterion => ({ criterionId: criterion.id, status: "satisfied", evidenceRefs: [] })), + remainingWork: [] }, evidence: [], verification: [], artifacts: [], attentionRequests: [] })); + const server = createServer(); + registerLocalRuntimeApiServer(server); + await new Promise(resolveListen => server.listen(0, "127.0.0.1", resolveListen)); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("TCP listener required"); + setupRunnerPrpWebSocketServer(server, { apiUrl: `http://127.0.0.1:${address.port}` }); + const bundle = createRunnerdCodexTransport({ runnerBinary: defaultCapabilityRunnerdBinary(), + codexCommand: resolve(import.meta.dirname, "../../../../packages/paperclip-runner/runner/target/debug/fake-codex-app-server"), + codexArgs: ["--state-file", join(input.root, "provider.json"), "--terminal-result-file", report, "--call-log", calls, "--record-process-start"], + stateDirectory: input.root, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null }, + runtimeContext: execution.runtimeContext, runnerRuntimeContext: execution.runtimeContext, + sourceCodexHome: home, environment: { CODEX_HOME: home, PATH: process.env.PATH }, + prpIdentity: { runnerInstanceId: native.runnerInstanceId, environmentLeaseId: native.environmentLeaseId, + runId: scheduled.id, normalizedSessionId: native.normalizedSessionId, turnId: native.turnId, itemId: native.itemId }, + controlPlaneRegistration: authority => registerRunnerPrpAuthority({ companyId: scheduled.companyId, runId: scheduled.id, authority }), + }); + const backend = createNativeSessionBackend(execution, { runnerInstanceId: native.runnerInstanceId, + environment: { CODEX_HOME: home, PATH: process.env.PATH }, codexTransportFactory: () => bundle.transport }); + try { + expect(await executePaperclipNativeSession({ db: input.db, execution, backend, useRunnerd: true, + runnerInstanceId: native.runnerInstanceId, turnTimeoutMs: 15_000 })).toMatchObject({ exitCode: 0, timedOut: false }); + await workspace.finalize(); + await reconcileNativeFinalizations(input.db, [scheduled.id]); + const [finished] = await input.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, scheduled.id)); + expect(finished, JSON.stringify(finished.failureDetail)).toMatchObject({ phase: "committed", attempt: 2 }); + expect(await input.db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, scheduled.id))) + .toMatchObject([{ schemaStatus: "accepted" }]); + expect((await readFile(calls, "utf8")).split("\n").filter(line => line === "process-start")).toHaveLength(1); + expect(await readFile(join(workspace.cwd, "saved.txt"), "utf8")).toBe("work survives a failed provider session\n"); + } finally { + await bundle.transport.close(); runnerPrpWebSocketInternals.resetForTests(); + server.closeAllConnections(); await new Promise(resolveClose => server.close(() => resolveClose())); + } +} diff --git a/server/src/__tests__/helpers/native-drain-controller.test-worker.ts b/server/src/__tests__/helpers/native-drain-controller.test-worker.ts new file mode 100644 index 00000000000..424d7598a24 --- /dev/null +++ b/server/src/__tests__/helpers/native-drain-controller.test-worker.ts @@ -0,0 +1,50 @@ +import { createServer } from "node:http"; +import { closeRegisteredClients, createDb, heartbeatRunEvents } from "@paperclipai/db"; +import { registerLocalRuntimeApiServer } from "../../runtime-api.js"; +import { setStartupRecoveryPhase } from "../../startup-recovery-state.js"; +import { heartbeatService } from "../../services/heartbeat.js"; +import { executePaperclipNativeSession } from "../../services/native-runtime/native-session-executor.js"; + +process.send?.({ ready: true }); +process.once("message", async (input: { + connectionString: string; + options: Omit[0], "db">; + providerEvidence?: boolean; +}) => { + const api = createServer((_request, response) => response.writeHead(404).end()); + registerLocalRuntimeApiServer(api); + let commands = 0; + let errorName: string | undefined; + let errorMessage: string | undefined; + try { + await new Promise(resolve => api.listen(0, "127.0.0.1", resolve)); + const db = createDb(input.connectionString); + try { + await executePaperclipNativeSession({ ...input.options, db, + runnerExecutionTarget: { ...input.options.runnerExecutionTarget!, + runner: { execute: async () => { commands++; throw new Error("Drained controller launched remote work"); } }, + } as never, + onEvent: async event => { + if (event.payload?.span !== "native.coordinator.claim") return; + if (input.providerEvidence) await db.insert(heartbeatRunEvents).values({ + companyId: input.options.execution.binding.companyId, agentId: input.options.execution.binding.agentId, + runId: input.options.execution.binding.runId, seq: 1_000_000, eventType: "turn.started", + sourceEventId: "controlled-provider-evidence" }); + setStartupRecoveryPhase("draining"); + await new Promise((resolve, reject) => api.close(error => error ? reject(error) : resolve())); + }, + }); + } catch (error) { + errorName = error instanceof Error ? error.name : String(error); + errorMessage = error instanceof Error ? error.message : String(error); + } + const heartbeat = heartbeatService(db); + try { + await heartbeat.drainRunningRunsForShutdown("SIGTERM", new Date(), [input.options.execution.binding.runId]); + } finally { await heartbeat.drainActiveRunExecutions(); } + await closeRegisteredClients(input.connectionString, { scope: "database" }); + process.send?.({ done: true, commands, errorName, errorMessage }, () => process.exit(0)); + } catch (error) { + process.send?.({ error: error instanceof Error ? error.stack : String(error) }, () => process.exit(1)); + } +}); diff --git a/server/src/__tests__/status-card-notifications.test.ts b/server/src/__tests__/status-card-notifications.test.ts index 0566ebdd394..7afd070f973 100644 --- a/server/src/__tests__/status-card-notifications.test.ts +++ b/server/src/__tests__/status-card-notifications.test.ts @@ -25,7 +25,7 @@ it.each([false, true])("keeps execution queue consumers intact and emits only a const write = vi.fn(async () => "written"); expect(await writeExecutionControl(owner as never, DELIVERY_QUEUES.workspaceRecovery, write, scope)).toBe("written"); expect(write).toHaveBeenCalledOnce(); - expect(new Set(settled.map(row => row.queue))).toEqual(new Set([DELIVERY_QUEUES.issueMonitors, DELIVERY_QUEUES.workspaceRecovery, + expect(new Set(settled.map(row => row.queue))).toEqual(new Set([DELIVERY_QUEUES.executionReplacement, DELIVERY_QUEUES.issueMonitors, DELIVERY_QUEUES.workspaceRecovery, ...(scoped ? [DELIVERY_QUEUES.statusCardEvaluation] : [])])); } finally { for (const stop of unsubscribe) stop(); } }); diff --git a/server/src/index.ts b/server/src/index.ts index 13c55b20807..0ba89a9fd11 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -81,7 +81,7 @@ import { embeddedPostgresOwnerPort } from "./embedded-postgres-owner.js"; import { deliverExecutionStatuses } from "./services/execution-status-delivery.js"; import { deliverReconciledExecutions, settleUnrecoverableExecutions } from "./services/execution-recovery-resolution.js"; import { deliverPendingPauseResumes } from "./services/issue-tree-resume.js"; -import { reconcileSafeNativeReplacements } from "./services/native-runtime/native-safe-replacement.js"; +import { invalidateNativeReplacementDiscovery, reconcileSafeNativeReplacements } from "./services/native-runtime/native-safe-replacement.js"; import { reconcileAbandonedExecutionControl } from "./services/execution-control-reconciliation.js"; import { acknowledgeNativeStartupCleanup } from "./services/native-runtime/native-startup-cleanup.js"; import { executionControlSchedule, registerExecutionControlWork } from "./services/execution-control-work.js"; @@ -1344,7 +1344,7 @@ async function startServerWithDatabaseTeardown( { queue: DELIVERY_QUEUES.executionStatus, run: () => deliverExecutionStatuses(db), next: executionSchedule.status }, { queue: DELIVERY_QUEUES.executionDisposition, run: () => settleUnrecoverableExecutions(db), next: executionSchedule.disposition }, ...(heartbeat ? [ - { queue: DELIVERY_QUEUES.executionReplacement, run: () => reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession: run => verifyStoppedNativeSessionForReplacement(db, run) }), next: executionSchedule.replacement }, + { queue: DELIVERY_QUEUES.executionReplacement, run: () => reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession: run => verifyStoppedNativeSessionForReplacement(db, run) }), next: executionSchedule.replacement, onSettled: () => invalidateNativeReplacementDiscovery(db), settleEmptyWhilePaused: true }, { queue: DELIVERY_QUEUES.executionContinuation, run: async () => { await deliverReconciledExecutions(db, heartbeat.wakeup); await deliverPendingPauseResumes(db, heartbeat.wakeup); diff --git a/server/src/services/delivery-work-coordinator.ts b/server/src/services/delivery-work-coordinator.ts index 39df770662d..97dc7a8ea87 100644 --- a/server/src/services/delivery-work-coordinator.ts +++ b/server/src/services/delivery-work-coordinator.ts @@ -25,7 +25,7 @@ export function createDeliveryWorkCoordinator(input: { // do not postpone the obligation indefinitely. coalesceMs?: number; run: (signal: AbortSignal) => Promise; - hasPending: () => Promise; + hasPending: (runAllowed?: boolean) => Promise; // Scoped hints belong to the outer transaction, including uncertainty recovery. onSettled?: (hints?: ReadonlySet) => void; // Optional maintenance/retry deadline restored from durable state. @@ -75,7 +75,7 @@ export function createDeliveryWorkCoordinator(input: { let runFailed = false; controller = attempt; running = withDatabaseOperation(`delivery.${queue}`, () => Promise.resolve().then(() => reconcileDatabaseWork(input.owner, queue)) - .then(() => workerStopped || !(input.canRun() || (task.runDuringDrain && input.canReconcile?.())) ? undefined : task.run(attempt.signal).catch(error => { runFailed = true; throw error; })).then(() => workerStopped ? false : task.hasPending()).then(pending => { + .then(() => workerStopped || !(input.canRun() || (task.runDuringDrain && input.canReconcile?.())) ? undefined : task.run(attempt.signal).catch(error => { runFailed = true; throw error; })).then(() => workerStopped ? false : task.hasPending(input.canRun() || Boolean(task.runDuringDrain && input.canReconcile?.()))).then(pending => { pendingWork = pending || databaseWorkPending(input.owner, queue); if (pendingWork) schedule(task.retryMs); else { finishIntent?.(); finishIntent = undefined; } diff --git a/server/src/services/environment-cleanup-notifications.ts b/server/src/services/environment-cleanup-notifications.ts index fbbf3352359..6beca498981 100644 --- a/server/src/services/environment-cleanup-notifications.ts +++ b/server/src/services/environment-cleanup-notifications.ts @@ -31,7 +31,10 @@ export async function notifyEnvironmentCleanup(transaction: object, scope?: Envi await notifyNativeEnvironmentSettlement(transaction, scope); await notifyQueuedRunDispatch(transaction, scope); // A settled run lease can make that run's abandoned native owner releasable. - if (scope?.runId) await notifyDeliveryWork(transaction, DELIVERY_QUEUES.executionFinalization); + if (scope?.runId) { + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.executionFinalization); + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.executionReplacement); + } await notifyDeliveryWork(transaction, DELIVERY_QUEUES.environmentCleanup); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.historicalCleanup); } diff --git a/server/src/services/execution-control-work.ts b/server/src/services/execution-control-work.ts index 7cedc0c2ea3..5a0e24f24e0 100644 --- a/server/src/services/execution-control-work.ts +++ b/server/src/services/execution-control-work.ts @@ -5,7 +5,7 @@ import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator. import { type DeliveryQueue } from "./delivery-work-notifications.js"; import { EXECUTION_RECONCILIATION_INTERVAL_MS } from "./execution-control-deadline.js"; import { obsoleteExecutionConversationHold, unresolvedExecutionDispositions } from "./execution-recovery-resolution.js"; -import { nativeReplacementPending } from "./native-runtime/native-safe-replacement.js"; +import { hasPendingNativeReplacements } from "./native-runtime/native-safe-replacement.js"; import { nextAbandonedNativeOwnerReleaseAt } from "./native-runtime/native-failed-run-owner-release.js"; import { hasPendingPauseResume } from "./issue-tree-resume.js"; @@ -16,6 +16,7 @@ export function registerExecutionControlWork( run: () => Promise; next: () => Promise; onSettled?: () => void; + settleEmptyWhilePaused?: boolean; }>, canRun: () => boolean = () => true, ) { @@ -24,6 +25,7 @@ export function registerExecutionControlWork( let reconciled = 0; let next: number | null = null; let earliestRunAt = 0; + let pausedGeneration = 0, pausedPending = true; const worker = coordinator.register(duty.queue, { retryMs: EXECUTION_RECONCILIATION_INTERVAL_MS, pausePassiveMaintenance: true, @@ -42,7 +44,18 @@ export function registerExecutionControlWork( next = deadline === null ? null : Math.max(deadline, earliestRunAt); reconciled = generation; }, - hasPending: async () => reconciled !== requested, + hasPending: async (runAllowed = true) => { + if (reconciled === requested) return false; + if (runAllowed || !duty.settleEmptyWhilePaused || !canRun() || reconciled === 0) return true; + if (pausedGeneration !== requested) { + const generation = requested; + const deadline = await duty.next(); + if (generation !== requested) return true; + pausedGeneration = generation; pausedPending = deadline !== null; next = deadline; + if (!pausedPending) reconciled = generation; + } + return pausedPending; + }, nextRunAt: () => next, }); return { ...worker, wake: () => { requested += 1; worker.wake(); } }; @@ -74,10 +87,7 @@ export function executionControlSchedule(db: Db) { return at === null ? null : deadline([new Date(at)]); }, async replacement() { - const rows = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns) - .innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id)) - .where(nativeReplacementPending()).limit(1); - return rows.length ? retry() : null; + return await hasPendingNativeReplacements(db) ? retry() : null; }, async continuation() { const rows = await db.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and( diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts index 0a9ed4725cc..756410efc6a 100644 --- a/server/src/services/explicit-native-continuation.test.ts +++ b/server/src/services/explicit-native-continuation.test.ts @@ -1586,7 +1586,7 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes await persistHeartbeatRunProcessMetadata(db, source.id, { pid: 999999999, processGroupId: null, startedAt: new Date().toISOString() }); await db.update(heartbeatRuns).set({ processPid: null }).where(eq(heartbeatRuns.id, source.id)); expect(await admit(f, true)).toBeNull(); - expect(await recordNativeLocalProcessStop(db, source)).toBe(true); + expect(await db.transaction(tx => recordNativeLocalProcessStop(tx as unknown as typeof db, source))).toBe(true); await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId: source.id, agentId: f.agentId, eventType: PROCESS_START_REQUESTED }); // No PID was stored for the new launch, as when the server dies after spawn. @@ -2440,7 +2440,7 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes await db.update(issues).set({ conversationAgentId: f.agentId, conversationUserId: "board", conversationState: "active", conversationBoundaryCommentId: boundaryId }).where(eq(issues.id, f.issueId)); await db.update(issueRecoveryActions).set({ createdAt: new Date("2026-09-11T09:00:00Z") }).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId)); - expect(await recordNativeLocalProcessStop(db, source)).toBe(true); + expect(await db.transaction(tx => recordNativeLocalProcessStop(tx as unknown as typeof db, source))).toBe(true); await db.update(heartbeatRuns).set({ processPid: null }).where(eq(heartbeatRuns.id, f.sourceRunId)); if (scenario === "after_reset_workspace_hold") { await db.update(heartbeatRuns).set({ processPid: 999999999, resultJson: { workspaceRestoreFailure: "restore_unsafe_archive" } }).where(eq(heartbeatRuns.id, f.sourceRunId)); diff --git a/server/src/services/native-local-process-stop.ts b/server/src/services/native-local-process-stop.ts index 37fc1569498..85d10f95359 100644 --- a/server/src/services/native-local-process-stop.ts +++ b/server/src/services/native-local-process-stop.ts @@ -6,6 +6,7 @@ import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js"; import { nativeSessionProviderEvidence } from "./native-runtime/native-session-resume.js"; import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; import os from "node:os"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; export const PROCESS_START_REQUESTED = "native.process_start_requested"; export const PROCESS_IDENTITY_RECORDED = "native.process_identity_recorded"; @@ -56,6 +57,8 @@ export async function recordNativeLocalProcessStop(db: Db, run: Pick lease.provider !== "local")) return false; if ((run.processPid && !absent(run.processPid)) || (run.processGroupId && !absent(-run.processGroupId))) return false; + await notifyDeliveryWork(db, DELIVERY_QUEUES.executionReplacement, + JSON.stringify({ companyId: run.companyId, runId: run.id })); await appendHeartbeatRunEvent(db, { companyId: run.companyId, runId: run.id, diff --git a/server/src/services/native-runtime/native-case-handoff.test.ts b/server/src/services/native-runtime/native-case-handoff.test.ts index 599b1e5584b..c13f5241e7c 100644 --- a/server/src/services/native-runtime/native-case-handoff.test.ts +++ b/server/src/services/native-runtime/native-case-handoff.test.ts @@ -84,7 +84,7 @@ describe("native persisted case handoff runtime", () => { expect(await isCaseHandoffSourceReleased(db, f.input)).toBe(false); const exited = once(child, "exit"); child.kill("SIGTERM"); await exited; const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); - expect(await recordNativeLocalProcessStop(db, run!)).toBe(true); + expect(await db.transaction(tx => recordNativeLocalProcessStop(tx as unknown as typeof db, run!))).toBe(true); // A new worker has no process-local state and reconstructs the release. expect(await isCaseHandoffSourceReleased(db, f.input)).toBe(true); expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]?.resultJson).toHaveProperty("caseHandoffExecutionReleased"); diff --git a/server/src/services/native-runtime/native-runner-restart-preparation.integration.test.ts b/server/src/services/native-runtime/native-runner-restart-preparation.integration.test.ts index 20eff3138be..350eee86282 100644 --- a/server/src/services/native-runtime/native-runner-restart-preparation.integration.test.ts +++ b/server/src/services/native-runtime/native-runner-restart-preparation.integration.test.ts @@ -1,11 +1,21 @@ import { startNativeSessionExecutionLeaseRenewal, type NativeExecutionLeaseRenewal } from "./native-execution-lease.js"; import { randomUUID } from "node:crypto"; -import { resolve } from "node:path"; +import { resolve, join } from "node:path"; +import { mkdir, readFile, writeFile } from "node:fs/promises"; import { eq, sql } from "drizzle-orm"; -import { environmentLeases, environments, heartbeatRuns, nativeRunFinalizations, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db"; +import { environmentLeases, environments, heartbeatRuns, nativeRunFinalizations, issueRecoveryActions, issues, completionContracts, nativeRunResults, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db"; import { beforeAll, expect, it, vi } from "vitest"; -import { DurablePrpControlPlane } from "../../vendor/paperclip-runner/index.js"; +import { spawn } from "node:child_process"; +import { createRequire } from "node:module"; +import { DurablePrpControlPlane, createNativeSessionBackend, createRunnerdCodexTransport, defaultCapabilityRunnerdBinary, canonicalNativeRuntimeContextDigest } from "../../vendor/paperclip-runner/index.js"; import { buildNativeExecutionInput } from "./native-execution-input.js"; +import { buildNativeCompletionContract } from "./completion-contracts.js"; +import { reconcileNativeFinalizations, claimNativeSessionResumptions } from "./native-finalization-reconciler.js"; +import { prepareNativeWorkspaceSync } from "./native-workspace-sync.js"; +import { workspaceOperationService } from "../workspace-operations.js"; +import type { EnvironmentLease } from "@paperclipai/shared"; +import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime"; +import { materializeAsset } from "./runtime-context.js"; import { nativeRuntimeContextFixture } from "./runtime-context.test-fixture.js"; import { executePaperclipNativeSession } from "./native-session-executor.js"; import { registerNativeRunnerCheckpoint } from "./native-runner-checkpoint.js"; @@ -21,6 +31,147 @@ describeEmbeddedPostgres("native runner restart recovery: preparation", () => { let successor: typeof environment.current.successor; beforeAll(() => { ({ runtimeRoot, companyId, agentId, successor } = environment.current); }); + it.each(["recover", "provider event", "cancelled task"] as const)("reconciles the drained remote backend with %s", async mode => { + const fixture = await seedRun(`DRAINED-BACKEND-${mode.replaceAll(" ", "-")}`); + const environmentId = randomUUID(), leaseId = randomUUID(); + await fixture.db.insert(environments).values({ id: environmentId, name: `Drained backend ${environmentId}`, driver: "sandbox" }); + const [lease] = await fixture.db.insert(environmentLeases).values({ id: leaseId, companyId, environmentId, + heartbeatRunId: fixture.runId, issueId: fixture.issueId, status: "active", + provider: "fixture", providerLeaseId: `drained-backend-${leaseId}` }).returning(); + const [issue] = await fixture.db.select().from(issues).where(eq(issues.id, fixture.issueId)); + const [contract] = await fixture.db.select().from(completionContracts).where(eq(completionContracts.issueId, fixture.issueId)); + const runtimeContext = nativeRuntimeContextFixture(); + runtimeContext.instructions.bundle = await materializeAsset([{ path: "AGENTS.md", mode: 0o444, + content: Buffer.from("Complete the controlled restart fixture task.\n") }]); + runtimeContext.aggregateDigest = canonicalNativeRuntimeContextDigest(runtimeContext); + const sandbox = join(runtimeRoot, `workspace-${fixture.runId}`); + await mkdir(sandbox, { recursive: true }); + const savedFile = join(sandbox, "saved-before-drain.txt"); + await writeFile(savedFile, "Preserve admitted work.\n"); + const execution = buildNativeExecutionInput({ companyId, agentId, runId: fixture.runId, + issue: { id: fixture.issueId, identifier: "NRR-DRAINED", title: "Continue after drain", description: null, workMode: "standard" }, + taskPrompt: "Continue the admitted work once.", provider: "codex", + workspace: { id: fixture.runId, cwd: sandbox, repoUrl: null, repoRef: null, branchName: null }, + normalizedSessionId: fixture.native.normalizedSessionId, runtimeContext, + completionContract: { id: contract.id, sha256: contract.canonicalSha256, schemaVersion: "paperclip.completion-contract.v1", + contract: buildNativeCompletionContract(issue, { revision: contract.revision }) }, + }); + await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: { + ...fixture.preparedProfile, nativeExecutionInput: execution, + } }).where(eq(heartbeatRuns.id, fixture.runId)); + const remoteSandbox = join(runtimeRoot, `remote-${fixture.runId}`); + const shell: CommandManagedRuntimeRunner = { execute: async input => { + const startedAt = new Date().toISOString(); + const child = spawn(input.command, input.args ?? [], { cwd: input.cwd, + env: { ...process.env, ...input.env }, stdio: ["pipe", "pipe", "pipe"] }); + let stdout = "", stderr = "", timedOut = false; + child.stdout.setEncoding("utf8").on("data", chunk => { stdout += chunk; }); + child.stderr.setEncoding("utf8").on("data", chunk => { stderr += chunk; }); + child.stdin.on("error", () => undefined); + child.stdin.end(input.stdin); + const timeout = setTimeout(() => { timedOut = true; child.kill("SIGKILL"); }, input.timeoutMs ?? 30_000); + try { return await new Promise((resolveResult, reject) => { + child.once("error", reject); + child.once("close", (exitCode, signal) => resolveResult({ stdout, stderr, exitCode, signal, timedOut, + pid: child.pid ?? null, startedAt })); + }); } finally { clearTimeout(timeout); } + } }; + const workspaceSync = await prepareNativeWorkspaceSync({ db: fixture.db, runId: fixture.runId, companyId, + workspaceId: fixture.runId, workspaceLocalDir: sandbox, lease: lease as unknown as EnvironmentLease, + target: { kind: "remote", transport: "sandbox", environmentId, leaseId, remoteCwd: remoteSandbox, runner: shell } }); + expect(workspaceSync).not.toBeNull(); + const child = spawn(process.execPath, ["--import", createRequire(import.meta.url).resolve("tsx"), + resolve(import.meta.dirname, "../../__tests__/helpers/native-drain-controller.test-worker.ts")], + { stdio: ["ignore", "pipe", "pipe", "ipc"], env: process.env }); + const logs: string[] = []; + child.stdout?.on("data", data => logs.push(String(data))); + child.stderr?.on("data", data => logs.push(String(data))); + const exited = new Promise(resolveExit => child.once("exit", resolveExit)); + try { + const result = await new Promise<{ commands: number; errorName: string; errorMessage: string }>((resolveResult, reject) => { + const deadline = setTimeout(() => reject(new Error(`Controller handoff timed out: ${logs.join("").slice(-2000)}`)), 60_000); + child.once("error", error => { clearTimeout(deadline); reject(error); }); + child.once("exit", code => { clearTimeout(deadline); reject(new Error(`Controller exited before handoff (${code}): ${logs.join("").slice(-2000)}`)); }); + child.on("message", (message: { ready?: boolean; done?: boolean; commands: number; errorName: string; errorMessage: string; error?: string }) => { + if (message.ready) child.send({ connectionString: environment.current.temporary.connectionString, providerEvidence: mode === "provider event", + options: { execution, runnerInstanceId: fixture.native.runnerInstanceId, useRunnerd: true, + runnerExecutionTarget: { kind: "remote", transport: "sandbox", environmentId, leaseId, remoteCwd: remoteSandbox } } }); + if (message.done) { clearTimeout(deadline); resolveResult(message); } + if (message.error) { clearTimeout(deadline); reject(new Error(message.error)); } + }); + }); + expect(result).toMatchObject({ commands: 0, errorMessage: "local_runtime_api_listener_unavailable" }); + expect(await exited).toBe(0); + const [handoff] = await fixture.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId)); + if (mode === "provider event") { + expect(handoff).toMatchObject({ phase: "terminal_failure", nextAttemptAt: null, + failureDetail: { recoveryMode: "ambiguous_state", providerSessionEstablished: true } }); + expect(await claimNativeSessionResumptions({ db: fixture.db, runnerInstanceId: "no-provider-replay", + runIds: [fixture.runId], deferIneligible: false })).toHaveLength(0); + return; + } + expect(handoff).toMatchObject({ phase: "retryable_failure", attempt: 1, + failureDetail: { originalFailureCode: "local_runtime_api_listener_unavailable", recoveryMode: "bootstrap_retry", + providerSessionEstablished: false, providerEventsExist: false, checkpointExists: false } }); + if (mode === "cancelled task") { + await fixture.db.update(issues).set({ status: "cancelled" }).where(eq(issues.id, fixture.issueId)); + } + const claims = (await Promise.all(["successor-a", "successor-b"].map(runnerInstanceId => + claimNativeSessionResumptions({ db: fixture.db, runnerInstanceId, runIds: [fixture.runId], + now: new Date(handoff.nextAttemptAt!.getTime() + 1), deferIneligible: false }), + ))).flat(); + if (mode === "cancelled task") { expect(claims).toHaveLength(0); return; } + expect(claims).toHaveLength(1); + const [run] = await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)); + expect(run).toMatchObject({ status: "running", nativeSessionId: fixture.native.normalizedSessionId }); + const [action] = await fixture.db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, fixture.issueId)); + expect(action).toMatchObject({ ownerType: "agent", ownerAgentId: agentId }); + if (process.env.PAPERCLIP_TEST_DRAIN_CONTINUATION === "1") { + const claim = claims[0]; + const root = join(runtimeRoot, `continued-${fixture.runId}`); + const home = join(root, "codex-home"); + await mkdir(home, { recursive: true }); + const report = join(root, "result.json"); + const calls = join(root, "calls.log"); + await writeFile(report, JSON.stringify({ schema: "paperclip.run_result.v1", reportedWorkDisposition: "done", + summary: "Admitted work completed after controller restart.", completionClaim: { + contractRevision: execution.completionContract.contract.revision, objectiveSatisfied: true, + criteria: execution.completionContract.contract.criteria.map(criterion => ({ criterionId: criterion.id, + status: "satisfied", evidenceRefs: [] })), remainingWork: [] }, + evidence: [], verification: [], artifacts: [], attentionRequests: [] })); + const bundle = createRunnerdCodexTransport({ ...transportOptions(fixture, root), + runnerBinary: defaultCapabilityRunnerdBinary(), + codexArgs: ["--state-file", join(root, "provider.json"), "--terminal-result-file", report, + "--call-log", calls, "--record-process-start"], + runtimeContext: execution.runtimeContext, runnerRuntimeContext: execution.runtimeContext, + sourceCodexHome: home, environment: { CODEX_HOME: home, PATH: process.env.PATH }, + lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null }, + }); + const backend = createNativeSessionBackend(execution, { runnerInstanceId: fixture.native.runnerInstanceId, + environment: { CODEX_HOME: home, PATH: process.env.PATH }, codexTransportFactory: () => bundle.transport }); + try { + expect(await executePaperclipNativeSession({ db: fixture.db, execution, backend, useRunnerd: true, + runnerInstanceId: fixture.native.runnerInstanceId, leaseOwner: claim.leaseOwner, + runnerExecutionTarget: { kind: "remote", transport: "sandbox", environmentId, leaseId, remoteCwd: remoteSandbox, + runner: { execute: async () => { throw new Error("Unexpected remote preparation after the controlled backend owns execution"); } } as never }, + turnTimeoutMs: 15_000 })).toMatchObject({ exitCode: 0, timedOut: false }); + const recorder = workspaceOperationService(fixture.db).createRecorder({ companyId, + issueId: fixture.issueId, heartbeatRunId: fixture.runId }); + const saved = await recorder.recordOperation({ phase: "workspace_finalize", cwd: sandbox, + run: async () => { await workspaceSync!.restoreWorkspace(); return { status: "succeeded", exitCode: 0 }; } }); + expect(saved.status).toBe("succeeded"); + await reconcileNativeFinalizations(fixture.db, [fixture.runId]); + expect(await fixture.db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, fixture.runId))) + .toMatchObject([{ schemaStatus: "accepted" }]); + const [finished] = await fixture.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId)); + expect(finished, JSON.stringify(finished.failureDetail)).toMatchObject({ phase: "committed", attempt: 2 }); + expect(await readFile(savedFile, "utf8")).toBe("Preserve admitted work.\n"); + expect((await readFile(calls, "utf8")).split("\n").filter(line => line === "process-start")).toHaveLength(1); + } finally { await bundle.transport.close(); } + } + } finally { child.kill("SIGKILL"); await exited; await workspaceSync?.cleanup(); } + }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); + it.each(["preparation", "claim lock"])("gives a fresh remote claim its full lease through blocked %s", async (blockedPhase) => { const fixture = await seedRun(`FRESH-SLOW-${blockedPhase === "claim lock" ? "CLAIM-LOCK" : "PREPARATION"}`); const environmentId = randomUUID(), leaseId = randomUUID(); diff --git a/server/src/services/native-runtime/native-safe-replacement.test.ts b/server/src/services/native-runtime/native-safe-replacement.test.ts index 491bae17b2b..088ae1bffc6 100644 --- a/server/src/services/native-runtime/native-safe-replacement.test.ts +++ b/server/src/services/native-runtime/native-safe-replacement.test.ts @@ -16,6 +16,11 @@ import { deliverReconciledExecutions, } from "../execution-recovery-resolution.js"; import { randomUUID } from "node:crypto"; +import { spawn } from "node:child_process"; +import { persistHeartbeatRunProcessMetadata } from "../heartbeat/run-lifecycle.js"; +import { hasNativeLocalProcessStop, releaseSettledNativeRunProcess } from "../native-local-process-stop.js"; +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import { executionControlSchedule } from "../execution-control-work.js"; import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; import { tmpdir } from "node:os"; import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; @@ -26,6 +31,7 @@ import { agents, companies, createDb, + subscribeDatabaseWork, heartbeatRunEvents, heartbeatRuns, issueRecoveryActions, @@ -41,7 +47,7 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, } from "../../__tests__/helpers/embedded-postgres.js"; -import { reconcileSafeNativeReplacements } from "./native-safe-replacement.js"; +import { invalidateNativeReplacementDiscovery, reconcileSafeNativeReplacements } from "./native-safe-replacement.js"; import { reconcileAbandonedExecutionControl } from "../execution-control-reconciliation.js"; const externalDatabaseUrl = process.env.PAPERCLIP_TEST_DATABASE_URL; const support = externalDatabaseUrl @@ -219,6 +225,103 @@ const support = externalDatabaseUrl } }, ); + it("wakes a completed replacement scan after actual local stop proof commits", async () => { + const source = await seed(2); + const workspace = await mkdtemp(join(tmpdir(), "paperclip-late-stop-")); + const child = spawn(process.execPath, ["-e", "process.exit(0)"], { detached: true, stdio: "ignore" }); + const pid = child.pid!; + await new Promise((resolve, reject) => { child.once("exit", () => resolve()); child.once("error", reject); }); + await persistHeartbeatRunProcessMetadata(db, source.runId, { pid, processGroupId: pid, + startedAt: new Date().toISOString(), targetKind: "local" }); + await db.update(heartbeatRuns).set({ finishedAt: new Date(), runnerProfileJson: { + nativeExecutionInput: { provider: { kind: "codex" }, workspace: { cwd: workspace } }, + } }).where(eq(heartbeatRuns.id, source.runId)); + await db.update(nativeRunFinalizations).set({ failureCode: "provider_transport_failed" }) + .where(eq(nativeRunFinalizations.runId, source.runId)); + await appendHeartbeatRunEvent(db, { companyId: source.companyId, runId: source.runId, agentId: source.agentId, + eventType: "harness.ready", stream: "system", message: "Provider started before its transport failed." }); + const verifyStoppedSession = async () => await hasNativeLocalProcessStop(db, source.companyId, source.runId) + ? { evidence: {}, retire: () => true } : null; + invalidateNativeReplacementDiscovery(db); + expect(await reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession })).toMatchObject({ scheduled: 0 }); + expect(await executionControlSchedule(db).replacement()).toBeNull(); + let wakes = 0; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.executionReplacement, event => { + if (event === "settled") { wakes++; invalidateNativeReplacementDiscovery(db); } + }); + try { + await db.transaction(async tx => { + expect(await releaseSettledNativeRunProcess(tx as unknown as typeof db, source.runId)).toBe("released"); + expect(wakes).toBe(0); + }); + expect(wakes).toBeGreaterThan(0); + expect(await executionControlSchedule(db).replacement()).not.toBeNull(); + expect(await reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession })).toMatchObject({ scheduled: 1 }); + expect(await reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession })).toMatchObject({ scheduled: 0 }); + expect(await executionControlSchedule(db).replacement()).toBeNull(); + const successors = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, source.runId)); + expect(successors).toHaveLength(1); + } finally { unsubscribe(); await rm(workspace, { recursive: true, force: true }); } + }); + it("wakes completed recovery when the real task becomes eligible", async () => { + const source = await seed(); + const workspace = await mkdtemp(join(tmpdir(), "paperclip-late-eligibility-")); + await db.update(heartbeatRuns).set({ runnerProfileJson: { recoveryEventInventoryVersion: 1, + nativeExecutionInput: { provider: { kind: "codex" }, workspace: { cwd: workspace } }, + } }).where(eq(heartbeatRuns.id, source.runId)); + await issueService(db).update(source.issueId, { status: "backlog" }); + invalidateNativeReplacementDiscovery(db); + expect(await reconcileSafeNativeReplacements(db)).toMatchObject({ scheduled: 0 }); + expect(await executionControlSchedule(db).replacement()).toBeNull(); + let wakes = 0; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.executionReplacement, event => { + if (event === "settled") { wakes++; invalidateNativeReplacementDiscovery(db); } + }); + try { + await issueService(db).update(source.issueId, { status: "in_progress" }); + expect(wakes).toBeGreaterThan(0); + expect(await executionControlSchedule(db).replacement()).not.toBeNull(); + expect(await reconcileSafeNativeReplacements(db)).toMatchObject({ scheduled: 1 }); + expect(await reconcileSafeNativeReplacements(db)).toMatchObject({ scheduled: 0 }); + expect(await executionControlSchedule(db).replacement()).toBeNull(); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, source.runId))).toHaveLength(1); + } finally { unsubscribe(); await rm(workspace, { recursive: true, force: true }); } + }); + it("advances past a failed full page and retries it after later work progresses", async () => { + const workspace = await mkdtemp(join(tmpdir(), "paperclip-failed-page-")); + const sources: Awaited>[] = []; + for (let index = 0; index < 26; index++) { + const source = await seed(); sources.push(source); + await db.update(heartbeatRuns).set({ runnerProfileJson: { + nativeExecutionInput: { provider: { kind: "codex" }, workspace: { cwd: workspace } }, + } }).where(eq(heartbeatRuns.id, source.runId)); + await db.update(nativeRunFinalizations).set({ failureCode: "provider_transport_failed" }) + .where(eq(nativeRunFinalizations.runId, source.runId)); + } + sources.sort((a, b) => a.runId.localeCompare(b.runId)); + const later = sources.at(-1)!; + let failing = true; + const verifyStoppedSession = vi.fn(async (run: typeof heartbeatRuns.$inferSelect) => { + if (failing && run.id !== later.runId) throw new Error("Controlled stopped-session verifier failure"); + return { evidence: {}, retire: () => true }; + }); + invalidateNativeReplacementDiscovery(db); + try { + await expect(reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession })) + .rejects.toThrow("native_replacement_decision_retry_required"); + expect(verifyStoppedSession).toHaveBeenCalledTimes(25); + expect(await executionControlSchedule(db).replacement()).not.toBeNull(); + expect(await reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession })).toMatchObject({ scheduled: 1 }); + expect(verifyStoppedSession).toHaveBeenCalledTimes(26); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, later.runId))).toHaveLength(1); + failing = false; + expect(await reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession })).toMatchObject({ scheduled: 25 }); + expect(await reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession })).toMatchObject({ scheduled: 0 }); + expect(await executionControlSchedule(db).replacement()).toBeNull(); + const successors = await db.select().from(heartbeatRuns).where(inArray(heartbeatRuns.retryOfRunId, sources.map(source => source.runId))); + expect(successors).toHaveLength(26); + } finally { await rm(workspace, { recursive: true, force: true }); } + }, 60_000); it.each(["unproven", "changed", "verified"] as const)("requires stopped-session proof through commit (%s)", async (mode) => { const source = await seed(2); await db.update(nativeRunFinalizations).set({ failureCode: "native_session_cleanup_quarantined" }).where(eq(nativeRunFinalizations.runId, source.runId)); @@ -235,8 +338,11 @@ const support = externalDatabaseUrl eventType: "tool.execution.started", stream: "system", payload: { name: "paperclip_finish", executionId: "completion", transport: "process" }, }); - await reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession }); - await reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession }); + for (let sweep = 0; sweep < 2; sweep++) { + const decision = reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession }); + if (mode === "changed") await expect(decision).rejects.toThrow("native_replacement_decision_retry_required"); + else await decision; + } const successors = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, source.runId)); expect(successors).toHaveLength(mode === "verified" ? 1 : 0); const [hold] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, source.issueId)); diff --git a/server/src/services/native-runtime/native-safe-replacement.ts b/server/src/services/native-runtime/native-safe-replacement.ts index 89ce98400e5..b2ce3bef1bc 100644 --- a/server/src/services/native-runtime/native-safe-replacement.ts +++ b/server/src/services/native-runtime/native-safe-replacement.ts @@ -5,7 +5,7 @@ import { notifyExecutionRunWork } from "../execution-control-notifications.js"; import { logger } from "../../middleware/logger.js"; import { createHash, randomUUID } from "node:crypto"; import { stat } from "node:fs/promises"; -import { and, eq, inArray, isNull, sql } from "drizzle-orm"; +import { and, asc, eq, gt, inArray, isNull, or, sql } from "drizzle-orm"; import { agentWakeupRequests, environmentLeases, @@ -27,6 +27,8 @@ import { issueRecoveryActionService } from "../issue-recovery-actions.js"; import { buildExecutionContinuation } from "../execution-continuation.js"; import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; import { hasRemoteTerminationReceipt } from "../remote-execution-termination.js"; +import { NATIVE_PROVIDER_DURABLE_EVENT_TYPES } from "./native-retired-backup-source.js"; +import { completedWorkspaceRecoveryCondition } from "../workspace-recovery-receipt.js"; export const NATIVE_SAFE_REPLACEMENT_REASON = "native_safe_replacement"; const record = (v: unknown): Record => @@ -45,6 +47,66 @@ const processAlive = (pid: number | null) => { type ReplacementDb = Db | Parameters[0]>[0]; +const listenerUnavailable = "local_runtime_api_listener_unavailable"; +function historicalListenerGuardPending() { + const profile = heartbeatRuns.runnerProfileJson, detail = nativeRunFinalizations.failureDetail; + const providerEvents = sql.join([...NATIVE_PROVIDER_DURABLE_EVENT_TYPES].map(value => sql`${value}`), sql`, `); + return and( + eq(nativeRunFinalizations.failureCode, "native_session_interrupted"), + eq(heartbeatRuns.errorCode, "native_session_interrupted"), eq(heartbeatRuns.error, listenerUnavailable), + sql`${detail}->>'message' = ${listenerUnavailable}`, + sql`${detail}->>'originalFailureCode' = 'native_session_interrupted'`, + sql`${detail}->>'recoveryMode' = 'ambiguous_state'`, + sql`${detail}->'providerSessionEstablished' = 'false'::jsonb`, + sql`${detail}->'providerEventsExist' = 'false'::jsonb`, + sql`${detail}->'checkpointExists' = 'false'::jsonb`, + eq(nativeRunFinalizations.attempt, 1), + isNull(nativeRunFinalizations.leaseOwner), isNull(nativeRunFinalizations.leaseExpiresAt), + isNull(nativeRunFinalizations.assessmentId), isNull(nativeRunFinalizations.decisionId), + isNull(heartbeatRuns.processPid), isNull(heartbeatRuns.processGroupId), + isNull(heartbeatRuns.sessionIdBefore), isNull(heartbeatRuns.sessionIdAfter), isNull(heartbeatRuns.retryOfRunId), + sql`not (coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) ?| array['cancellation', 'nativeCancellation'])`, + sql`coalesce(${profile}->'sessionCheckpoint', 'null'::jsonb) = 'null'::jsonb`, + sql`coalesce(${profile}->'nativeProviderPackAdmission', 'null'::jsonb) = 'null'::jsonb`, + sql`coalesce(${profile}#>>'{nativeProviderPackAdmissionState,state}', 'pending') = 'pending'`, + sql`${profile}#>>'{nativeExecutionInput,binding,runId}' = ${heartbeatRuns.id}::text`, + sql`${profile}#>>'{nativeExecutionInput,binding,companyId}' = ${heartbeatRuns.companyId}::text`, + sql`${profile}#>>'{nativeExecutionInput,binding,agentId}' = ${heartbeatRuns.agentId}::text`, + sql`${profile}#>>'{nativeExecutionInput,binding,issueId}' = ${nativeRunFinalizations.issueId}::text`, + sql`${profile}#>>'{nativeExecutionInput,session,normalizedSessionId}' = ${heartbeatRuns.nativeSessionId}::text`, + sql`not exists (select 1 from ${heartbeatRuns} prior where prior.company_id = ${heartbeatRuns.companyId} + and prior.native_session_id = ${heartbeatRuns.nativeSessionId} and prior.id <> ${heartbeatRuns.id})`, + sql`not exists (select 1 from ${heartbeatRunEvents} event where event.company_id = ${heartbeatRuns.companyId} + and event.run_id = ${heartbeatRuns.id} and (event.source_event_id is not null + or event.event_type in (${providerEvents}) or event.event_type ~ '^(turn|item|tool|semantic_tool|run[.]result)[.]'))`, + sql`not exists (select 1 from ${toolInvocations} invocation where invocation.company_id = ${heartbeatRuns.companyId} + and invocation.run_id = ${heartbeatRuns.id})`, + sql`not exists (select 1 from ${issueRecoveryActions} + where ${issueRecoveryActions.companyId} = ${heartbeatRuns.companyId} + and ${issueRecoveryActions.sourceIssueId} = ${nativeRunFinalizations.issueId} + and ${issueRecoveryActions.cause} = 'native_continuation_requires_reconciliation' + and ${issueRecoveryActions.evidence}->>'runId' = ${heartbeatRuns.id}::text + and ${issueRecoveryActions.evidence}->'automaticRecovery'->>'replay' = 'blocked' + and not coalesce((${completedWorkspaceRecoveryCondition(heartbeatRuns.id)}), false))`, + sql`exists (select 1 from ${issues} task where task.company_id = ${heartbeatRuns.companyId} + and task.id = ${nativeRunFinalizations.issueId} and task.assignee_agent_id = ${heartbeatRuns.agentId} + and task.status in ('todo', 'in_progress', 'in_review', 'blocked') + and (task.status <> 'blocked' or exists (select 1 from ${issueRecoveryActions} hold + where hold.company_id = task.company_id and hold.source_issue_id = task.id and hold.kind = 'active_run_watchdog' + and hold.cause = ${nativeRunFinalizations.failureCode} and hold.evidence->>'runId' = ${heartbeatRuns.id}::text + and hold.evidence#>>'{nativeFailureBlock,runId}' = ${heartbeatRuns.id}::text + and hold.evidence#>'{nativeFailureBlock,statusVersion}' = to_jsonb(task.status_version))))`, + ); +} + +async function historicalListenerGuardEvidence(db: ReplacementDb, run: typeof heartbeatRuns.$inferSelect) { + const [evidence] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns) + .innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id)).where(and( + eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.id, run.id), historicalListenerGuardPending(), + )).limit(1); + return Boolean(evidence); +} + /** A saved remote source is proved by its current ledger, never a controller path. */ async function remoteWorkspaceReplacementEvidence( db: ReplacementDb, @@ -113,16 +175,95 @@ export function nativeReplacementPending() { return and( eq(heartbeatRuns.status, "failed"), eq(nativeRunFinalizations.phase, "terminal_failure"), - inArray( + or(inArray( nativeRunFinalizations.failureCode, ["native_provider_terminal_failed", "native_session_cleanup_quarantined", "provider_transport_failed"], - ), + ), historicalListenerGuardPending()), isNull(nativeRunFinalizations.resultId), sql`coalesce(${nativeRunFinalizations.failureDetail}->>'successorRunId', '') = ''`, sql`coalesce(${nativeRunFinalizations.failureDetail}->>'replacementDenied', '') = ''`, ); } +const REPLACEMENT_SCAN_LIMIT = 100; +const replacementDiscovery = new WeakMap(); +function replacementDiscoveryState(db: Db) { + let state = replacementDiscovery.get(db); + if (!state) { state = { after: undefined, complete: false, generation: 0, rescan: false }; replacementDiscovery.set(db, state); } + return state; +} + +export function invalidateNativeReplacementDiscovery(db: Db) { + const state = replacementDiscoveryState(db); + state.complete = false; state.rescan = true; state.generation++; +} + +async function replacementHistoryReady(db: Db, run: typeof heartbeatRuns.$inferSelect, issueId: string, retryDatabaseErrors = false) { + try { + const continuation = await buildExecutionContinuation({ db, companyId: run.companyId, issueId, + agentId: run.agentId, context: record(run.contextSnapshot), summary: null, exposeLowTrustRaw: false }); + return continuation.unresolvedInteractionIds.length === 0; + } catch (error) { + let cause: unknown = error; + for (let depth = 0; retryDatabaseErrors && depth < 4 && cause && typeof cause === "object"; depth++) { + const failure = cause as { code?: unknown; cause?: unknown }; + if (typeof failure.code === "string" && /^[A-Z0-9_]+$/.test(failure.code)) throw error; + cause = failure.cause; + } + return null; + } +} + +async function discoverNativeReplacements(db: Db, after: string | undefined, limit = 25) { + const candidates: Array<{ run: typeof heartbeatRuns.$inferSelect; coordinator: typeof nativeRunFinalizations.$inferSelect }> = []; + let scanned = 0; + for (;;) { + const rows = await db.select({ run: heartbeatRuns, coordinator: nativeRunFinalizations }).from(heartbeatRuns) + .innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id)) + .where(and(nativeReplacementPending(), after ? gt(heartbeatRuns.id, after) : undefined)) + .orderBy(asc(heartbeatRuns.id)).limit(Math.min(25, REPLACEMENT_SCAN_LIMIT - scanned)); + if (!rows.length) return { candidates, scanned, after: undefined }; + for (const row of rows) { + after = row.run.id; scanned++; + let ready = true; + if (row.coordinator.failureCode === "native_session_interrupted") { + ready = false; + if (row.run.runnerProfileJson?.remoteWorkspace != null) { + const leases = await db.select().from(environmentLeases).where(and( + eq(environmentLeases.companyId, row.run.companyId), eq(environmentLeases.heartbeatRunId, row.run.id), + )); + if (leases.every(lease => lease.releasedAt !== null)) { + const evidence = await remoteWorkspaceReplacementEvidence(db, row.run, row.coordinator.issueId, leases); + ready = Boolean(evidence?.workspacePreserved && evidence.providerStopped) + && await replacementHistoryReady(db, row.run, row.coordinator.issueId, true) === true; + } + } + } + if (ready) candidates.push(row); + if (candidates.length >= limit || scanned >= REPLACEMENT_SCAN_LIMIT) return { candidates, scanned, after }; + } + if (rows.length < 25) return { candidates, scanned, after: undefined }; + } +} + +function settleReplacementDiscovery(state: ReturnType, after: string | undefined, generation: number) { + const rescan = state.rescan || generation !== state.generation; + state.after = after; state.complete = !after && !rescan; + state.rescan = Boolean(after && rescan); +} + +export async function hasPendingNativeReplacements(db: Db) { + const state = replacementDiscoveryState(db); + if (state.complete) return false; + if (state.after) return true; + state.rescan = false; + const generation = state.generation; + const result = await discoverNativeReplacements(db, undefined, 1); + if (result.candidates.length) return true; + settleReplacementDiscovery(state, result.after, generation); + return !state.complete; +} + /** Retryable outbox: terminal_failure plus a new-format failure code is the durable candidate. */ export async function reconcileSafeNativeReplacements( db: Db, @@ -136,18 +277,14 @@ export async function reconcileSafeNativeReplacements( failpoint?: (phase: "successor_inserted" | "lineage_committed") => void; } = {}, ) { - const candidates = await db - .select({ run: heartbeatRuns, coordinator: nativeRunFinalizations }) - .from(heartbeatRuns) - .innerJoin( - nativeRunFinalizations, - eq(nativeRunFinalizations.runId, heartbeatRuns.id), - ) - .where( - nativeReplacementPending(), - ) - .limit(25); - let scheduled = 0; + const discovery = replacementDiscoveryState(db); + if (discovery.complete) discovery.after = undefined; + discovery.complete = false; + if (!discovery.after) discovery.rescan = false; + const generation = discovery.generation; + const selection = await discoverNativeReplacements(db, discovery.after); + const candidates = selection.candidates; + let scheduled = 0, pendingFailures = 0; for (const { run, coordinator } of candidates) { try { if ( @@ -170,11 +307,13 @@ export async function reconcileSafeNativeReplacements( ); // Teardown is still in progress. The next sweep rechecks its durable outcome. if (leases.some((lease) => lease.releasedAt === null)) continue; - const stoppedSession = coordinator.failureCode !== "native_provider_terminal_failed" + const historicalListener = coordinator.failureCode === "native_session_interrupted"; + if (historicalListener && !await historicalListenerGuardEvidence(db, run)) continue; + const stoppedSession = coordinator.failureCode !== "native_provider_terminal_failed" && !historicalListener ? await options.verifyStoppedSession?.(run) ?? null : null; // A transport label alone is not evidence. Keep inspecting these candidates // as process cleanup and the final transcript become durable. - if (coordinator.failureCode !== "native_provider_terminal_failed" && !stoppedSession) continue; + if (coordinator.failureCode !== "native_provider_terminal_failed" && !historicalListener && !stoppedSession) continue; const invocations = await db .select() .from(toolInvocations) @@ -263,23 +402,11 @@ export async function reconcileSafeNativeReplacements( ? [String(p.kind ?? p.type)] : []; }); - let historyComplete = false; - try { - const continuation = await buildExecutionContinuation({ - db, - companyId: run.companyId, - issueId: coordinator.issueId, - agentId: run.agentId, - context: record(run.contextSnapshot), - summary: null, - exposeLowTrustRaw: false, - }); - if (continuation.unresolvedInteractionIds.length > 0) continue; - historyComplete = true; - } catch { - /* The named operator outcome retains incomplete context. */ - } + const history = await replacementHistoryReady(db, run, coordinator.issueId); + if (history === false) continue; + const historyComplete = history === true; const remoteWorkspace = await remoteWorkspaceReplacementEvidence(db, run, coordinator.issueId, leases); + if (historicalListener && (!remoteWorkspace?.workspacePreserved || !remoteWorkspace.providerStopped || invocations.length)) continue; const decision = decideNativeReplacement({ failedSession: true, // A facade transport failure can conceal an authoritative protocol @@ -413,6 +540,13 @@ export async function reconcileSafeNativeReplacements( current.attempt >= 3 ) return false; + let historicalRun: typeof heartbeatRuns.$inferSelect | undefined; + if (historicalListener) { + const [currentRun] = await tx.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId), + )).for("update"); + historicalRun = currentRun; + } if (task.status === "blocked") { const failureHolds = await tx.select().from(issueRecoveryActions).where(and( eq(issueRecoveryActions.companyId, run.companyId), @@ -427,6 +561,14 @@ export async function reconcileSafeNativeReplacements( }); if (!ownsBlock) return false; } + const preservationHolds = historicalListener ? await tx.select().from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.companyId, run.companyId), eq(issueRecoveryActions.sourceIssueId, task.id), + eq(issueRecoveryActions.kind, "active_run_watchdog"), + eq(issueRecoveryActions.cause, "native_continuation_requires_reconciliation"), + sql`${issueRecoveryActions.evidence}->>'runId' = ${run.id}`, + sql`${issueRecoveryActions.evidence}->'automaticRecovery'->>'replay' = 'blocked'`, + )).for("update") : []; + if (historicalListener && (!historicalRun || !await historicalListenerGuardEvidence(tx, historicalRun))) return false; if (remoteWorkspace) { const currentWorkspace = await remoteWorkspaceReplacementEvidence(tx, run, coordinator.issueId, leases, true); if (!currentWorkspace?.workspacePreserved || !currentWorkspace.providerStopped) return false; @@ -548,6 +690,7 @@ export async function reconcileSafeNativeReplacements( eq(issueRecoveryActions.cause, current.failureCode!), sql`${issueRecoveryActions.evidence}->>'runId' = ${run.id}`, )).for("update"); + holds.push(...preservationHolds); if (holds.some(hold => hold.status === "active" || hold.status === "escalated")) { await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: run.companyId, issueId: task.id }); } @@ -587,11 +730,17 @@ export async function reconcileSafeNativeReplacements( } } catch (error) { if (options.failpoint) throw error; + pendingFailures++; logger.warn( { runId: run.id }, "Native replacement remains pending; another sweep will retry its durable decision", ); } } - return { scanned: candidates.length, scheduled }; + settleReplacementDiscovery(discovery, selection.after, generation); + if (pendingFailures) { + discovery.complete = false; discovery.rescan = true; + throw new Error("native_replacement_decision_retry_required"); + } + return { scanned: selection.scanned, scheduled }; } diff --git a/server/src/services/remote-workspaces.integration.test.ts b/server/src/services/remote-workspaces.integration.test.ts index 912416f1457..9e45c05470b 100644 --- a/server/src/services/remote-workspaces.integration.test.ts +++ b/server/src/services/remote-workspaces.integration.test.ts @@ -13,7 +13,7 @@ import { pathToFileURL } from "node:url"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { and, eq, sql } from "drizzle-orm"; import { activityLog, agents, agentTaskSessions, agentWakeupRequests, applyPendingMigrations, closeRegisteredClients, companies, createDb, environmentLeases, environments, - executionWorkspaceRepositories, executionWorkspaces, executionWorkspaceSnapshots, heartbeatRunEvents, heartbeatRuns, issueRecoveryActions, issues, livenessDb, nativeRunFinalizations, projects, projectWorkspaces, workspaceOperations } from "@paperclipai/db"; + subscribeDatabaseWork, executionWorkspaceRepositories, executionWorkspaces, executionWorkspaceSnapshots, heartbeatRunEvents, heartbeatRuns, issueRecoveryActions, issueThreadInteractions, issues, livenessDb, nativeRunFinalizations, projects, projectWorkspaces, workspaceOperations } from "@paperclipai/db"; import type { AdapterExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime"; import { buildProjectMentionHref, type Environment, type EnvironmentLease } from "@paperclipai/shared"; @@ -25,12 +25,13 @@ import { createSavedWorkspaceLoader, prepareRemoteWorkspace, readRemoteWorkspace clearRemoteWorkspaceLocationForReleasedLease, createRemoteWorkspaceGitAuth, prepareRemoteWorkspaceRepository, reserveRemoteWorkspace, saveRemoteWorkspace, shouldUseRemoteWorkspaces } from "./remote-workspaces.js"; import type { createGitRemoteAuthProvider } from "./git-credentials.js"; -import { reconcileSafeNativeReplacements } from "./native-runtime/native-safe-replacement.js"; +import { invalidateNativeReplacementDiscovery, reconcileSafeNativeReplacements } from "./native-runtime/native-safe-replacement.js"; import { remoteTerminationReceipt } from "./remote-execution-termination.js"; import { RemoteWorkspaceRestoreError } from "./remote-workspace-transport.js"; import { readRemoteRunnerState } from "./native-runtime/native-session-executor.js"; import { notFound } from "../errors.js"; import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js"; +import { issueThreadInteractionService } from "./issue-thread-interactions.js"; import { createRemoteWorkspaceFiles } from "./remote-workspace-files.js"; import { workspaceFileResourceService } from "./workspace-file-resources.js"; import { nativeWorkspaceExportPendingForLease, preserveRemoteWorkspacePreparationFailure } from "./native-runtime/native-workspace-finalization-state.js"; @@ -47,6 +48,10 @@ import { instanceSettingsService } from "./instance-settings.js"; import { resolveRemoteWorkspaceSource } from "./heartbeat/remote-workspace-source.js"; import { toExecutionWorkspace } from "./execution-workspaces.js"; import { recoveryPostgresSupport, startRecoveryPostgres } from "../__tests__/helpers/recovery-postgres.js"; +import { getExecutionBlocker } from "./execution-blocker.js"; +import { executionControlSchedule } from "./execution-control-work.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import { createSystemRecovery } from "./system-recovery-state.js"; const exec = promisify(execFile); let databaseUrl = process.env.PAPERCLIP_REMOTE_WORKSPACE_TEST_DATABASE_URL; @@ -82,6 +87,7 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem await db.insert(issues).values({ id: ids.issue, companyId: ids.company, title: "Use a remote workspace", status: "in_progress", assigneeAgentId: ids.agent }); await db.insert(heartbeatRuns).values({ id: ids.run, companyId: ids.company, agentId: ids.agent, nativeIssueId: ids.issue, runtimeMode: "native", status: "running" }); + invalidateNativeReplacementDiscovery(db); await db.insert(environmentLeases).values({ id: ids.lease, companyId: ids.company, environmentId: ids.environment, heartbeatRunId: ids.run, issueId: ids.issue, provider: "daytona", providerLeaseId: `allocation-${ids.lease}`, status: "active" }); }); @@ -91,6 +97,10 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem ? (row.metadata.uploads as { objectKey: string }[]).map(item => item.objectKey) : [])].filter((value): value is string => !!value))) { await storage.deleteObject(ids.company, key); } + if (temporary && process.env.PAPERCLIP_TEST_HISTORICAL_LISTENER === "1") await db.transaction(async tx => { + await tx.execute(sql`select set_config('client_min_messages', 'warning', true)`); + await tx.execute(sql`TRUNCATE companies CASCADE`); + }); await db.delete(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.companyId, ids.company)); await db.delete(workspaceOperations).where(eq(workspaceOperations.companyId, ids.company)); await db.delete(activityLog).where(eq(activityLog.companyId, ids.company)); @@ -1712,14 +1722,20 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem } finally { await recovered.cleanup(); } }, 60_000); - it.each(["saved", "dirty", "unsaved", "foreign_snapshot", "delete_requested", "deletion_race", "controller_claim_race", "stale_epoch", "unconfirmed_stop", "unknown_action"] as const)( - "checks durable remote work at the native safe-replacement decision (%s)", async scenario => { + it.each(["saved", "dirty", "unsaved", "foreign_snapshot", "delete_requested", "deletion_race", "controller_claim_race", "stale_epoch", "unconfirmed_stop", "unknown_action", "historical_listener_guard", "historical_listener_provider_event", "historical_listener_checkpoint", "historical_listener_resumed", "historical_listener_other_message", "historical_listener_cancelled", "historical_listener_prior_session", "historical_listener_unconfirmed_stop", "historical_listener_dirty", "historical_listener_stale_block", "historical_listener_unknown_event", "historical_listener_budget", "historical_listener_bad_preservation", "historical_listener_absent", "historical_listener_later_candidate", "historical_listener_late_proof", "historical_listener_lock_timeout", "historical_listener_late_interaction", "historical_listener_interaction_paused", "historical_listener_own_question"] as const)( + "checks durable remote work at the native safe-replacement decision (%s)", async kind => { + const historicalListener = kind.startsWith("historical_listener_"); + const processAcceptance = kind === "historical_listener_guard" && process.env.PAPERCLIP_TEST_HISTORICAL_LISTENER === "1"; + const scenario = kind === "historical_listener_late_proof" ? "dirty" : kind === "historical_listener_unconfirmed_stop" ? "unconfirmed_stop" + : kind === "historical_listener_dirty" ? "dirty" : historicalListener ? "saved" : kind; + const shouldSchedule = scenario === "saved" && (!historicalListener || ["historical_listener_guard", "historical_listener_later_candidate", "historical_listener_lock_timeout"].includes(kind)); const reserved = await reserve({ kind: "empty" }), first = await target("native-replacement-sandbox"); const prepared = await prepareRemoteWorkspace({ db, storage, workspace: reserved, runId: ids.run, lease: await lease(), target: first, assertOwned: async () => undefined }); await fs.writeFile(path.join(prepared.remoteRoot, "saved.txt"), "work survives a failed provider session\n"); if (scenario !== "unsaved") await saveRemoteWorkspace({ db, storage, workspace: prepared.workspace, - runId: ids.run, lease: prepared.lease, target: first, assertOwned: async () => undefined }); + runId: ids.run, lease: prepared.lease, target: first, assertOwned: async () => undefined, + boundary: kind === "historical_listener_late_proof" ? "initial" : "final" }); const current = await workspace(reserved.id), authority = readRemoteWorkspaceAuthority(current.metadata)!; if (scenario === "dirty" || scenario === "stale_epoch") await db.update(executionWorkspaces).set({ metadata: { ...current.metadata, remoteWorkspace: { ...authority, ...(scenario === "dirty" ? { dirty: true, writerRunId: ids.run } @@ -1729,29 +1745,137 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem .where(eq(executionWorkspaceSnapshots.id, authority.snapshotId!)); if (scenario === "delete_requested") await db.update(executionWorkspaceSnapshots).set({ status: "delete_requested" }) .where(eq(executionWorkspaceSnapshots.id, authority.snapshotId!)); - const stopped = (await db.update(environmentLeases).set({ status: "released", releasedAt: new Date(), cleanupStatus: "success", + if (processAcceptance) { + const { recordHistoricalListenerFailure } = await import("../__tests__/helpers/historical-listener-continuation.js"); + await recordHistoricalListenerFailure({ db, connectionString: databaseUrl!, runId: ids.run, cwd: prepared.remoteRoot, + environmentId: ids.environment, leaseId: ids.lease }); + } + const stopped = kind === "historical_listener_late_proof" ? prepared.lease : (await db.update(environmentLeases).set({ status: "released", releasedAt: new Date(), cleanupStatus: "success", metadata: { ...prepared.lease.metadata, ...(scenario !== "unconfirmed_stop" ? { remoteExecutionTermination: remoteTerminationReceipt(prepared.lease, { providerLeaseId: prepared.lease.providerLeaseId, state: "destroyed", }), } : {}) }, }).where(eq(environmentLeases.id, ids.lease)).returning())[0]! as EnvironmentLease; - if (scenario === "saved") { + if (shouldSchedule) { await db.transaction(tx => clearRemoteWorkspaceLocationForReleasedLease({ db: tx, lease: stopped })); await fs.rm(prepared.remoteRoot, { recursive: true, force: true }); await expect(fs.stat(prepared.remoteRoot)).rejects.toThrow(); } const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, ids.run)); - await db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date(), runnerProfileJson: { + const sessionId = randomUUID(); + await db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date(), + ...(historicalListener && !processAcceptance ? { nativeSessionId: sessionId } : {}), + ...(historicalListener ? { error: "local_runtime_api_listener_unavailable", errorCode: "native_session_interrupted" } : {}), runnerProfileJson: { ...source!.runnerProfileJson, recoveryEventInventoryVersion: 1, - nativeExecutionInput: { provider: { kind: "codex" }, workspace: { cwd: prepared.remoteRoot } }, + nativeExecutionInput: processAcceptance ? source!.runnerProfileJson!.nativeExecutionInput : { provider: { kind: "codex" }, workspace: { cwd: prepared.remoteRoot }, + ...(historicalListener ? { binding: { runId: ids.run, companyId: ids.company, agentId: ids.agent, issueId: ids.issue }, + session: { normalizedSessionId: sessionId } } : {}) }, } }).where(eq(heartbeatRuns.id, ids.run)); - await db.insert(nativeRunFinalizations).values({ runId: ids.run, companyId: ids.company, issueId: ids.issue, - phase: "terminal_failure", attempt: 1, failureCode: "native_provider_terminal_failed", - failureDetail: { originalFailureCode: "fixture_checkpoint_unusable" } }); + if (!processAcceptance) await db.insert(nativeRunFinalizations).values({ runId: ids.run, companyId: ids.company, issueId: ids.issue, + phase: "terminal_failure", attempt: 1, failureCode: historicalListener ? "native_session_interrupted" : "native_provider_terminal_failed", + failureDetail: historicalListener ? { originalFailureCode: "native_session_interrupted", message: "local_runtime_api_listener_unavailable", + recoveryMode: "ambiguous_state", providerSessionEstablished: false, providerEventsExist: false, checkpointExists: false } + : { originalFailureCode: "fixture_checkpoint_unusable" } }); + if (historicalListener && !processAcceptance) { + const blocked = await issueService(db).update(ids.issue, { status: "blocked" }); + await db.insert(issueRecoveryActions).values({ companyId: ids.company, sourceIssueId: ids.issue, + kind: "active_run_watchdog", cause: "native_session_interrupted", ownerType: "board", status: "active", + fingerprint: createHash("sha256").update(`${ids.run}:native_session_interrupted`).digest("hex"), + evidence: { runId: ids.run, nativeFailureBlock: { runId: ids.run, statusVersion: blocked!.statusVersion } }, + nextAction: "Reconcile the original startup failure." }); + } + if (kind === "historical_listener_guard" || kind === "historical_listener_bad_preservation") { + const actionId = randomUUID(), preservationId = randomUUID(); + await db.insert(issueRecoveryActions).values({ id: actionId, companyId: ids.company, sourceIssueId: ids.issue, + kind: "active_run_watchdog", cause: "native_continuation_requires_reconciliation", ownerType: "system", status: "resolved", outcome: "blocked", + fingerprint: `native-workspace:${ids.run}`, evidence: { runId: ids.run, + workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [ids.lease] }, + automaticRecovery: { replay: "blocked" }, continuationDelivery: "invalidated", + systemRecovery: { ...createSystemRecovery(ids.run), state: "complete", preservationOperationIds: [preservationId] }, + workspaceRecovery: { schema: "paperclip.workspace-recovery.v1", runId: ids.run, status: "recovered", stage: "saving_receipts", attempt: 1, requestIds: [] }, + }, nextAction: "Files were preserved without replay." }); + await db.insert(workspaceOperations).values([{ id: preservationId, companyId: ids.company, heartbeatRunId: ids.run, issueId: ids.issue, + executionWorkspaceId: reserved.id, phase: "workspace_recovery", status: kind === "historical_listener_bad_preservation" ? "failed" : "succeeded", startedAt: new Date(), + metadata: { recoveryActionId: actionId, snapshotId: authority.snapshotId } }, + { companyId: ids.company, heartbeatRunId: ids.run, issueId: ids.issue, phase: "workspace_finalize", status: "succeeded", startedAt: new Date(), + metadata: { owningService: "workspace_recovery", preservationOperationIds: [preservationId] } }]); + } + if (kind === "historical_listener_absent") await db.update(heartbeatRuns).set({ + runnerProfileJson: sql`${heartbeatRuns.runnerProfileJson} - 'remoteWorkspace'`, + }).where(eq(heartbeatRuns.id, ids.run)); + if (kind === "historical_listener_stale_block") await issueService(db).update(ids.issue, { status: "blocked", actorUserId: "operator" }); + if (kind === "historical_listener_unknown_event") await db.insert(heartbeatRunEvents).values({ companyId: ids.company, runId: ids.run, + agentId: ids.agent, seq: 100, eventType: "future.provider.activity", sourceEventId: "source-provider-event" }); + if (kind === "historical_listener_budget") await db.update(nativeRunFinalizations).set({ attempt: 3 }).where(eq(nativeRunFinalizations.runId, ids.run)); + if (kind === "historical_listener_checkpoint") await db.update(heartbeatRuns).set({ runnerProfileJson: { + ...((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, ids.run)))[0]!.runnerProfileJson), sessionCheckpoint: { providerSessionId: "already-started" }, + } }).where(eq(heartbeatRuns.id, ids.run)); + if (kind === "historical_listener_resumed") await db.update(heartbeatRuns).set({ sessionIdBefore: "prior-session" }).where(eq(heartbeatRuns.id, ids.run)); + if (kind === "historical_listener_prior_session") await db.insert(heartbeatRuns).values({ id: randomUUID(), companyId: ids.company, + agentId: ids.agent, status: "succeeded", nativeSessionId: sessionId }); + if (kind === "historical_listener_other_message") await db.update(nativeRunFinalizations).set({ failureDetail: { + message: "provider reported local_runtime_api_listener_unavailable", originalFailureCode: "native_session_interrupted", recoveryMode: "ambiguous_state", + providerSessionEstablished: false, providerEventsExist: false, checkpointExists: false, + } }).where(eq(nativeRunFinalizations.runId, ids.run)); + if (kind === "historical_listener_cancelled") await db.update(issues).set({ status: "cancelled" }).where(eq(issues.id, ids.issue)); + if (kind === "historical_listener_provider_event") await appendHeartbeatRunEvent(db, { companyId: ids.company, runId: ids.run, + agentId: ids.agent, eventType: "turn.started", stream: "system" }); + const interactionWake = ["historical_listener_late_interaction", "historical_listener_interaction_paused", "historical_listener_own_question"].includes(kind); + const interactionId = randomUUID(); + if (interactionWake) { + const priorRunId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: priorRunId, companyId: ids.company, agentId: ids.agent, + nativeIssueId: ids.issue, status: "succeeded", contextSnapshot: { issueId: ids.issue } }); + await db.insert(issueThreadInteractions).values({ id: interactionId, companyId: ids.company, issueId: ids.issue, + kind: "request_confirmation", status: "pending", sourceRunId: kind === "historical_listener_own_question" ? ids.run : priorRunId, + payload: { version: 1, prompt: "An obsolete question from earlier work." } }); + if (kind === "historical_listener_interaction_paused") await issueService(db).update(ids.issue, { status: "backlog" }); + if (kind === "historical_listener_own_question") await appendHeartbeatRunEvent(db, { companyId: ids.company, runId: ids.run, + agentId: ids.agent, eventType: "tool.execution.started", stream: "system", + payload: { name: "request_confirmation", executionId: "source-question", transport: "process" } }); + } if (scenario === "unknown_action") await appendHeartbeatRunEvent(db, { companyId: ids.company, runId: ids.run, agentId: ids.agent, eventType: "tool.execution.started", stream: "system", payload: { name: "send_email", executionId: "unconfirmed-external-action", transport: "process" } }); + if (kind === "historical_listener_later_candidate") { + const sourceNumber = BigInt(`0x${ids.run.replaceAll("-", "")}`); + const parked = Array.from({ length: 100 }, (_, index) => { + const hex = (sourceNumber - BigInt(index + 1)).toString(16).padStart(32, "0"); + return { runId: `${hex.slice(0,8)}-${hex.slice(8,12)}-${hex.slice(12,16)}-${hex.slice(16,20)}-${hex.slice(20)}`, + issueId: randomUUID(), sessionId: randomUUID() }; + }); + await db.insert(issues).values(parked.map(row => ({ id: row.issueId, companyId: ids.company, + title: "Preserve unavailable source", status: "in_progress", assigneeAgentId: ids.agent }))); + await db.insert(heartbeatRuns).values(parked.map(row => ({ id: row.runId, companyId: ids.company, agentId: ids.agent, + nativeIssueId: row.issueId, nativeSessionId: row.sessionId, runtimeMode: "native", status: "failed", + errorCode: "native_session_interrupted", error: "local_runtime_api_listener_unavailable", runnerProfileJson: { + nativeExecutionInput: { binding: { runId: row.runId, companyId: ids.company, agentId: ids.agent, issueId: row.issueId }, + session: { normalizedSessionId: row.sessionId } }, + } }))); + await db.insert(nativeRunFinalizations).values(parked.map(row => ({ runId: row.runId, companyId: ids.company, issueId: row.issueId, + phase: "terminal_failure", attempt: 1, failureCode: "native_session_interrupted", failureDetail: { + message: "local_runtime_api_listener_unavailable", originalFailureCode: "native_session_interrupted", recoveryMode: "ambiguous_state", + providerSessionEstablished: false, providerEventsExist: false, checkpointExists: false, + } }))); + expect(await reconcileSafeNativeReplacements(db)).toMatchObject({ scanned: 100, scheduled: 0 }); + expect(await executionControlSchedule(db).replacement()).not.toBeNull(); + invalidateNativeReplacementDiscovery(db); + } + if (kind === "historical_listener_lock_timeout") { + let release!: () => void, announce!: () => void; + const locked = new Promise(resolve => { announce = resolve; }); + const released = new Promise(resolve => { release = resolve; }); + const holder = db.transaction(async tx => { + await tx.select({ id: issues.id }).from(issues).where(eq(issues.id, ids.issue)).for("update"); + announce(); await released; + }); + await locked; + try { + await expect(reconcileSafeNativeReplacements(db)).rejects.toThrow(); + expect(await executionControlSchedule(db).replacement()).not.toBeNull(); + } finally { release(); await holder; } + } let replacement; if (scenario === "deletion_race" || scenario === "controller_claim_race") { let release!: () => void, announce!: () => void; @@ -1783,12 +1907,102 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem expect(blocked).toBe(true); } finally { release(); await deletion; replacement = await pending; } } else replacement = await reconcileSafeNativeReplacements(db); - expect(replacement).toMatchObject({ scheduled: scenario === "saved" ? 1 : 0 }); + expect(replacement).toMatchObject({ scheduled: shouldSchedule ? 1 : 0 }); const successors = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, ids.run)); - expect(successors).toHaveLength(scenario === "saved" ? 1 : 0); - if (scenario === "saved") expect(successors[0]).toMatchObject({ status: "scheduled_retry", + expect(successors).toHaveLength(shouldSchedule ? 1 : 0); + if (shouldSchedule) expect(successors[0]).toMatchObject({ status: "scheduled_retry", contextSnapshot: { retryOfRunId: ids.run, forceFreshSession: true } }); - else expect(await fs.readFile(path.join(prepared.remoteRoot, "saved.txt"), "utf8")) + if (kind === "historical_listener_later_candidate") { + for (let pass = 0; pass < 3 && await executionControlSchedule(db).replacement() !== null; pass++) { + expect(await reconcileSafeNativeReplacements(db)).toMatchObject({ scheduled: 0 }); + } + expect(await executionControlSchedule(db).replacement()).toBeNull(); + } + if (kind === "historical_listener_guard") expect(await getExecutionBlocker(db, ids.company, ids.issue)).toBeNull(); + if (["historical_listener_provider_event", "historical_listener_checkpoint", "historical_listener_resumed", "historical_listener_other_message", + "historical_listener_cancelled", "historical_listener_prior_session", "historical_listener_unknown_event", "historical_listener_budget", "historical_listener_stale_block", "historical_listener_bad_preservation", "historical_listener_dirty", "historical_listener_unconfirmed_stop", "historical_listener_absent", "historical_listener_late_proof"].includes(kind)) { + expect(await executionControlSchedule(db).replacement()).toBeNull(); + } + if (interactionWake) { + expect(await executionControlSchedule(db).replacement()).toBeNull(); + let wakes = 0; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.executionReplacement, event => { + if (event === "settled") { wakes++; invalidateNativeReplacementDiscovery(db); } + }); + try { + await issueThreadInteractionService(db).skipInteraction({ id: ids.issue, companyId: ids.company, + status: kind === "historical_listener_interaction_paused" ? "backlog" : "blocked" }, interactionId, {}, { userId: "local-board" }); + expect(wakes).toBeGreaterThan(0); + expect(await reconcileSafeNativeReplacements(db)).toMatchObject({ scheduled: kind === "historical_listener_late_interaction" ? 1 : 0 }); + expect(await reconcileSafeNativeReplacements(db)).toMatchObject({ scheduled: 0 }); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, ids.run))) + .toHaveLength(kind === "historical_listener_late_interaction" ? 1 : 0); + expect(await executionControlSchedule(db).replacement()).toBeNull(); + } finally { unsubscribe(); } + } + if (kind === "historical_listener_late_proof") { + let wakes = 0; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.executionReplacement, event => { + if (event === "settled") { wakes++; invalidateNativeReplacementDiscovery(db); } + }); + try { + const saved = await saveRemoteWorkspace({ db, storage, workspace: await workspace(reserved.id), runId: ids.run, + lease: await lease(), target: first, assertOwned: async () => undefined }); + expect(wakes).toBeGreaterThan(0); + expect(await executionControlSchedule(db).replacement()).toBeNull(); + const priorWakes = wakes, active = await lease(), pluginId = randomUUID(); + await db.update(environmentLeases).set({ metadata: { ...active.metadata, driver: "sandbox", pluginId, + provider: "daytona", sandboxProviderPlugin: true } }).where(eq(environmentLeases.id, ids.lease)); + const runtime = environmentRuntimeService(db, { pluginWorkerManager: { + isRunning: () => true, getWorker: () => ({ supportedMethods: ["environmentReleaseLease"] }), + call: async (_id: string, method: string) => { + expect(method).toBe("environmentReleaseLease"); + return { providerLeaseId: active.providerLeaseId, state: "destroyed" }; + }, + } as never }); + await runtime.releaseRunLeases(ids.run, "failed"); + expect(wakes).toBeGreaterThan(priorWakes); + expect(await executionControlSchedule(db).replacement()).not.toBeNull(); + expect(await reconcileSafeNativeReplacements(db)).toMatchObject({ scheduled: 1 }); + expect(await reconcileSafeNativeReplacements(db)).toMatchObject({ scheduled: 0 }); + expect(await executionControlSchedule(db).replacement()).toBeNull(); + const restored = await createSavedWorkspaceLoader(db, storage)({ companyId: ids.company, workspaceId: reserved.id, snapshotId: saved.snapshotId }); + try { expect(await fs.readFile(path.join(restored.root, "saved.txt"), "utf8")).toBe("work survives a failed provider session\n"); } + finally { await restored.cleanup(); } + } finally { unsubscribe(); } + } + if (kind === "historical_listener_bad_preservation") { + expect(await getExecutionBlocker(db, ids.company, ids.issue)).toMatchObject({ cause: "native_continuation_requires_reconciliation" }); + + } + if (processAcceptance) { + expect(await reconcileSafeNativeReplacements(db)).toMatchObject({ scheduled: 0 }); + const { completeHistoricalListenerSuccessor } = await import("../__tests__/helpers/historical-listener-continuation.js"); + await completeHistoricalListenerSuccessor({ db, runId: successors[0]!.id, root: path.join(root, "successor-runtime"), + prepareWorkspace: async () => { + const leaseId = randomUUID(), runId = successors[0]!.id; + const [nextLease] = await db.insert(environmentLeases).values({ id: leaseId, companyId: ids.company, + environmentId: ids.environment, heartbeatRunId: runId, issueId: ids.issue, provider: "daytona", + providerLeaseId: `allocation-${leaseId}`, status: "active" }).returning(); + const next = await reserveRemoteWorkspace({ db, storage, companyId: ids.company, issueId: ids.issue, + runId, agentId: ids.agent, environmentId: ids.environment, projectId: null, projectWorkspaceId: null, + existingWorkspaceId: reserved.id, source: { kind: "empty" } }); + const nextTarget = { ...await target("historical-successor"), leaseId, sandboxId: nextLease.providerLeaseId! }; + const nextPrepared = await prepareRemoteWorkspace({ db, storage, workspace: next, runId, + lease: nextLease as EnvironmentLease, target: nextTarget }); + return { cwd: nextPrepared.remoteRoot, finalize: async () => { + const { workspaceOperationService } = await import("./workspace-operations.js"); + const recorder = workspaceOperationService(db).createRecorder({ companyId: ids.company, issueId: ids.issue, heartbeatRunId: runId }); + const saved = await recorder.recordOperation({ phase: "workspace_finalize", cwd: nextPrepared.remoteRoot, run: async () => { + await saveRemoteWorkspace({ db, storage, workspace: nextPrepared.workspace, runId, lease: nextPrepared.lease, + target: nextTarget, assertOwned: async () => undefined }); + return { status: "succeeded", exitCode: 0 }; + } }); + expect(saved.status).toBe("succeeded"); + } }; + } }); + } + if (!shouldSchedule) expect(await fs.readFile(path.join(prepared.remoteRoot, "saved.txt"), "utf8")) .toBe("work survives a failed provider session\n"); if (scenario === "unknown_action") { const [decision] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, ids.run)); diff --git a/server/src/services/remote-workspaces.ts b/server/src/services/remote-workspaces.ts index 63aa0bfe1e1..fbdd45e1e64 100644 --- a/server/src/services/remote-workspaces.ts +++ b/server/src/services/remote-workspaces.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { workspaceAllocationFiles } from "./remote-workspace-allocation.js"; import { readNativeWorkspaceMigration } from "./native-runtime/native-workspace-migration.js"; import { createHash, randomUUID } from "node:crypto"; @@ -763,7 +764,10 @@ export async function saveRemoteWorkspace(input: { eq(executionWorkspaceSnapshots.executionWorkspaceId, workspace.id), eq(executionWorkspaceSnapshots.status, "uploaded"))).returning({ id: executionWorkspaceSnapshots.id }); if (!committed) throw new Error("remote_workspace_snapshot_publication_changed"); await notifyRemoteWorkspaceSnapshots(tx, `workspace:${workspace.id}`); - if (boundary === "final") await notifyTerminalWorkspaceWork(tx, `workspace:${workspace.id}`); + if (boundary === "final") { + await notifyTerminalWorkspaceWork(tx, `workspace:${workspace.id}`); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionReplacement); + } const receipt = object(workspace.metadata?.remoteWorkspaceSnapshotRepositories); const priorRepositories = receipt.snapshotId === current.snapshotId && Array.isArray(receipt.repositories) ? remoteWorkspaceSourceSchema.parse({ kind: "empty", repositories: receipt.repositories }).repositories ?? [] : []; diff --git a/server/src/services/run-dispatch-notifications.ts b/server/src/services/run-dispatch-notifications.ts index 3cd4f00cbe1..36fad0e81a3 100644 --- a/server/src/services/run-dispatch-notifications.ts +++ b/server/src/services/run-dispatch-notifications.ts @@ -12,6 +12,7 @@ export async function notifyRunDispatch(transaction: object, scope?: RunDispatch await notifyIssueRecovery(transaction, scope ? { companyId: scope.companyId } : undefined); const hint = scope ? JSON.stringify({ companyId: scope.companyId }) : "*"; await notifyDeliveryWork(transaction, DELIVERY_QUEUES.nativeProcessSessions, hint); + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.executionReplacement, hint); await notifyQueuedRunDispatch(transaction, scope); } diff --git a/server/src/services/workspace-recovery.ts b/server/src/services/workspace-recovery.ts index 88f0efc82cc..b4d9a9d6595 100644 --- a/server/src/services/workspace-recovery.ts +++ b/server/src/services/workspace-recovery.ts @@ -390,6 +390,7 @@ export async function recoverRetainedWorkspace(input: Source & { db: Db; environ const activity = await db.transaction(async tx => { await notifyNativeResultCommit(tx, input); await notifyNativeWorkspaceFinalization(tx, input); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionReplacement); await notifyNativeStatusReassessment(tx, input); await tx.insert(workspaceOperations).values({ companyId: input.companyId, issueId: input.issueId, heartbeatRunId: input.runId, phase: "workspace_finalize", status: "succeeded", finishedAt: new Date(),