From e3fb4bc73ae412c91eb69800c60893a8e1486cab Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 6 Nov 2026 11:49:28 -0600 Subject: [PATCH] fix(native): recover saved listener startup failures (#19479) ## Thinking Path > - Paperclip manages AI agents and their work. > - Native recovery must preserve saved work and respect current task authority. > - A closed local API listener can stop startup before a provider starts. > - #19462 gives new failures a bounded bootstrap retry. > - Older saved failures still have a generic terminal code and a recovery hold. > - This pull request verifies the old producer outcome and uses the existing safe replacement path. > - Eligible tasks continue once. Unknown provider outcomes and user holds remain blocked. ## Linked Issues or Issue Description Refs: #19462 **What happened?** A restart can leave an old listener startup failure in `terminal_failure` with `native_session_interrupted`. Startup recovery does not claim it after #19462. A completed file recovery can also retain its no-replay hold. **Expected behavior** Continue automatically when durable records prove that no provider started, the source stopped, and the current files were saved. Keep the existing attempt budget and normal admission checks. **Steps to reproduce** 1. Close the actual API listener after a fresh native coordinator claim. 2. Persist the old executor's generic failure. Save the workspace and confirm source termination. 3. Restart recovery. The baseline schedules no successor. The repair schedules one successor after all receipts pass. **Paperclip version or commit** Baseline: `0252421d5c`. Old failure producer: `2f35eb1080`. **Deployment mode** Built from source. Local macOS arm64, Node 26.4.0, real embedded PostgreSQL, independent filesystem execution. No external AI service is used by the controlled acceptance. ## What Changed - Recognize the exact saved local-listener guard through one proof predicate shared by discovery and the locked transaction. - Require a fresh first attempt, matching execution bindings, no prior provider lineage, no provider events or tool calls, no checkpoint or admission receipt, and no cancellation. - Reuse current snapshot, exact stop, source block, attempt budget, and successor idempotency checks. - Wait for completed source file-recovery receipts. Retire only that same source's verified preservation hold in the successor transaction. - Bound discovery to 100 candidates per pass, keep completed history quiet, and wake through existing transactional producer notifications. - Retain retries after transient database and final ownership-proof errors; advance the bounded cursor and force a rescan so a failing page cannot hide later work. - Let only the replacement duty discharge a proved-empty notification during drain through its existing read-only readiness check, once per notification generation. Keep real work and concurrent changes outstanding without running replacement during drain. - Wake replacement at canonical local-stop proof writes and the existing dispatch eligibility boundary; these hints grant no new execution authority. - Add database regressions, scheduler quietness checks, real controller handoff coverage, and an opt-in controlled Runner continuation. ## Verification Final production head: `0df41b2fef2644195d1224d36984af3827da3e3b`. Local environment: macOS arm64, Node 26.4.0, TypeScript 7.0.2, Vitest 5.0.3, frozen-lockfile dependencies. Tests execute freshly; `--fsModuleCache` reuses only content-validated module transforms. - Native safe replacement: 62/62 PASS on isolated PostgreSQL 17 (63.18 seconds), including clean per-suite database disposal. All 62 assertions also passed on embedded PostgreSQL 18 beta, but its Darwin cleanup hook timed out; that invocation is not counted as a pass. Docker's disk was full, so the disposable PostgreSQL 17 data used a size-limited RAM mount, with no host mounts. The owned container was removed afterward. - Actual stopped-process producer suite: 31/31 PASS. A real exited child produces the stop receipt, and its replacement notification settles only after the outer transaction commits. - Workspace and separate-process drain regressions: 33/33 PASS (75.90 seconds). - Shared coordinator and execution-control lifecycle checks: 39/39 PASS (16.43 seconds). Pending work, concurrent changes, cold startup, shutdown, and passive drain boundaries remain covered. - Actual suppressed Cloud routine/drain workflow: PASS (34.64 seconds), using the unchanged test and timeouts. Its matching local RED reproduced CI's 462-query, zero-pending-query, one-idle-hold timeout after an admitted write committed during drain (69.46 seconds). - Actual historical controller handoff: PASS (48.40 seconds). A separate controller loads the original executor from `2f35eb1080`, writes its generic listener failure, and exits. Recovery at the final head admits exactly one successor. The real Rust Runner and controlled Codex process complete governed `paperclip_finish`; attempt two commits and the saved file remains intact. Only the child loads the old artifact, and both isolated artifacts are restored afterward. Rust source subtree `77961b70f177ec3c38d11abcaec92fd3d0e96ec0` matches the built binary. No external AI service is used. - Seven corrected existing fixtures pass: transactional stop-proof calls (5 cases) and notification inventory (2 cases). Their earlier CI failures exposed fixture calls outside the documented transaction and an outdated expected queue set. - Server typecheck: PASS. Red/green proofs also cover the original saved failure on baseline `0252421d5c` (zero successors instead of one), held-history quietness, bounded forward scanning through 100 parked rows and concurrent wakes, real snapshot/cleanup producer notifications, actual issue-row lock timeout, and task/history eligibility notifications. An obsolete prior-run question can be skipped to request another proof check; actual skip while the task is paused or the source owns provider tool activity schedules no successor. The failed-page regression starts with 25 persistent verifier errors and a healthy candidate 26. Baseline RED repeats the first page. Repair GREEN advances the healthy candidate, revisits the original failures after their fault clears, creates all 26 successors exactly once, and becomes quiet. Greptile is SUCCESS, 5/5 on the exact final head (2026-11-06 17:09:21 UTC). Final-head CI is GREEN: 56 successful checks, three intentional skips, one neutral result, and no pending or failing checks. [Attempt two and results](https://github.com/paperclipai/paperclip/actions/runs/42258352876/attempts/2) include passing build, typecheck, Rust, native server, package, browser E2E, serialized server, general server/chat tests, and final verification. Five jobs in the first attempt were interrupted together at 17:13:12 UTC; the native-server log records an explicit runner shutdown signal. The single infrastructure retry reran only those five jobs and dependent verification, preserving the 49 successful checks on the unchanged head. No timeout was weakened, and setup failures or interrupted invocations are not counted as passes. Focused commands from the repository root: ```sh ./server/node_modules/.bin/vitest run --project @paperclipai/server server/src/services/remote-workspaces.integration.test.ts server/src/services/native-runtime/native-runner-restart-preparation.integration.test.ts -t 'native safe-replacement decision|reconciles the drained remote backend' --fsModuleCache --reporter=dot ./server/node_modules/.bin/vitest run --project @paperclipai/server server/src/__tests__/execution-control-work.test.ts server/src/__tests__/delivery-work-coordinator.test.ts -t 'execution-control lifecycle|work coordinator' --fsModuleCache --reporter=dot node node_modules/typescript/bin/tsc -p server/tsconfig.json --noEmit --pretty false ``` The historical acceptance sets `PAPERCLIP_TEST_HISTORICAL_LISTENER=1` and uses an original-executor child artifact with a `try/finally` restore. Normal regressions require no Rust binary. ## Risks - Compatibility is deliberately limited to positively proved fresh remote startup failures. Other old ambiguous failures remain held. - Existing provider work, a changed task block, incomplete preservation, or revoked authority must prevent execution. Normal retry admission remains in control. - This PR does not deploy the change or mutate a hosted task. Hosted recovery still requires the repaired revision and all current proof gates. ## Model Used OpenAI GPT-6 through Codex, with reasoning, tool use, and code execution. The exact backend model identifier and context window are not exposed in this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/execution-semantics.md | 28 ++ .../__tests__/execution-control-work.test.ts | 35 +++ .../historical-listener-continuation.ts | 134 ++++++++++ .../native-drain-controller.test-worker.ts | 50 ++++ .../status-card-notifications.test.ts | 2 +- server/src/index.ts | 4 +- .../src/services/delivery-work-coordinator.ts | 4 +- .../environment-cleanup-notifications.ts | 5 +- server/src/services/execution-control-work.ts | 22 +- .../explicit-native-continuation.test.ts | 4 +- .../src/services/native-local-process-stop.ts | 3 + .../native-case-handoff.test.ts | 2 +- ...er-restart-preparation.integration.test.ts | 157 ++++++++++- .../native-safe-replacement.test.ts | 112 +++++++- .../native-runtime/native-safe-replacement.ts | 217 ++++++++++++--- .../remote-workspaces.integration.test.ts | 246 ++++++++++++++++-- server/src/services/remote-workspaces.ts | 6 +- .../services/run-dispatch-notifications.ts | 1 + server/src/services/workspace-recovery.ts | 1 + 19 files changed, 961 insertions(+), 72 deletions(-) create mode 100644 server/src/__tests__/helpers/historical-listener-continuation.ts create mode 100644 server/src/__tests__/helpers/native-drain-controller.test-worker.ts 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(),