diff --git a/packages/paperclip-runner/src/live/runnerd-turn-start-timeout-retention.test.ts b/packages/paperclip-runner/src/live/runnerd-turn-start-timeout-retention.test.ts new file mode 100644 index 00000000000..810cf8ec82e --- /dev/null +++ b/packages/paperclip-runner/src/live/runnerd-turn-start-timeout-retention.test.ts @@ -0,0 +1,102 @@ +import { execFileSync, spawn } from "node:child_process"; +import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { expect, it, vi } from "vitest"; +import { createRunnerdNativeSessionBackend } from "../backends/codex-native-backend.js"; +import type { NativeExecutionInputV1 } from "../contracts/native-execution.js"; +import type * as durableControlPlane from "../control-plane/durable-prp-control-plane.js"; +import type { PrpEvent } from "../protocol/replay-contract.js"; +import { executeNativeSession } from "../native-session-runtime.js"; +import { createCapabilityRunnerdCodexTransport } from "./runnerd-codex-transport.js"; +import { defaultCapabilityRunnerdBinary, fakeCodex, fakeCodexArgs } from "../../test/support/runnerd-transport.js"; + +function processState(pid: number): "alive" | "gone" { + try { process.kill(pid, 0); } catch { return "gone"; } + try { + return execFileSync("ps", ["-o", "stat=", "-p", String(pid)], { encoding: "utf8" }).trim().startsWith("Z") ? "gone" : "alive"; + } catch { return "gone"; } +} + +const kill = (pid: number | null | undefined) => { + if (!pid) return; + try { process.kill(pid, "SIGKILL"); } catch (error) { if (!["ESRCH", "EPERM"].includes(String((error as NodeJS.ErrnoException).code))) throw error; } +}; + +it("keeps the remote Runner alive for the same-run reattachment when the turn.start command times out", async () => { + const root = await mkdtemp(join(tmpdir(), "runnerd-turn-start-timeout-")); + const hangPath = join(root, "hang-turn-start"); + const identity = { runId: "run-start-timeout", sessionId: "session-start-timeout", companyId: "company-start-timeout", + issueId: "issue-start-timeout", agentId: "agent-start-timeout" }; + const execution: NativeExecutionInputV1 = { + schema: "paperclip.native-execution-input.v1", + binding: { runId: identity.runId, companyId: identity.companyId, issueId: identity.issueId, agentId: identity.agentId, + executionWorkspaceId: "workspace-start-timeout" }, + session: { normalizedSessionId: identity.sessionId, driverKind: "codex_app_server", protocolVersion: 1, + lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } }, + task: { identifier: "TEST-8", title: "Write the brief", description: null, prompt: "Write the brief.", workMode: "standard" }, + workspace: { cwd: root, repoUrl: null, repoRef: null, branchName: null }, + provider: { kind: "codex", model: null }, + completionContract: { id: "contract-start-timeout", sha256: "digest", schemaVersion: "paperclip.completion-contract.v1", contract: { + revision: "1", objective: "Write the brief", criteria: [{ id: "brief", requirement: "Write the brief" }], + } }, + interactionResponses: [], credentialBindings: [], + }; + const children: ReturnType[] = []; + const launch = vi.fn((spec: durableControlPlane.RunnerProcessLaunchSpec) => { + const child = spawn(spec.command, [...spec.args], { cwd: spec.cwd, env: spec.environment, stdio: "ignore" }); + children.push(child); + return { + child, + completion: new Promise((resolveExit, rejectExit) => { + child.once("error", rejectExit); + child.once("exit", (code, exitSignal) => resolveExit({ code, signal: exitSignal, stdout: "", stderr: "" })); + }), + isExitConfirmed: () => child.exitCode !== null || child.signalCode !== null, + }; + }); + const bundle = createCapabilityRunnerdCodexTransport({ + runnerBinary: defaultCapabilityRunnerdBinary(), codexCommand: fakeCodex, + codexArgs: fakeCodexArgs(root, "--hang-turn-start", hangPath), stateDirectory: root, + runnerReconnectGraceMs: 60_000, runnerProcessLauncher: launch, closeGraceMs: 600, turnStartTimeoutMs: 2_000, + prpIdentity: { runnerInstanceId: "runner-start-timeout", environmentLeaseId: "lease-start-timeout", runId: identity.runId, + normalizedSessionId: identity.sessionId, turnId: "turn-start-timeout", itemId: "item-start-timeout" }, + }); + const events: PrpEvent[] = []; + try { + await writeFile(hangPath, "before"); + const backend = createRunnerdNativeSessionBackend(execution, { runnerInstanceId: "runner-start-timeout", + transportFactory: () => bundle.transport }); + const executing = executeNativeSession({ + input: execution, backend, runnerInstanceId: "runner-start-timeout", controlPlaneInstanceId: "controller-start-timeout", + timeoutMs: 20_000, requireSessionCloseBeforeReturn: true, + controlPlane: { + async openRun() {}, + async checkpointSession() {}, + async appendEvent(event) { + if (!("sourceSeq" in event)) throw new Error("Expected a durable PRP event"); + events.push(event); + return { cursor: events.length, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed" }; + }, + async replayEvents(request) { + const source = events.filter(event => event.sourceInstanceId === request.sourceInstanceId); + return { events: source.filter(event => event.sourceSeq > request.afterSourceSeq), + highestContiguousSourceSeq: source.at(-1)?.sourceSeq ?? 0 }; + }, + async completeRun() {}, + }, + }); + await expect(executing).rejects.toThrow("provider_transport_failed: PRP command turn.start timed out"); + expect(launch).toHaveBeenCalledTimes(1); + const runner = children[0]!; + await new Promise(resolveWait => setTimeout(resolveWait, 1_000)); + expect(processState(runner.pid!)).toBe("alive"); + const journal = JSON.parse(await readFile(join(root, "runner", "runner-state.json"), "utf8")) as Record; + expect(journal.lifecycle).toBe("ready"); + } finally { + await rm(hangPath, { force: true }); + for (const child of children) kill(child.pid); + kill(bundle.evidence().codexPid); + await rm(root, { recursive: true, force: true }); + } +}, 60_000); diff --git a/server/src/services/runner-lost-mid-turn-continuation.test.ts b/server/src/services/runner-lost-mid-turn-continuation.test.ts index add97007f3c..73c942bf518 100644 --- a/server/src/services/runner-lost-mid-turn-continuation.test.ts +++ b/server/src/services/runner-lost-mid-turn-continuation.test.ts @@ -31,12 +31,18 @@ const stagingRefusal = (reason: "process_not_running" | "process_identity_change describe("a remote Runner that died in the middle of a provider turn", () => { useWorkspaceRecoveryFixture(targets); - async function runnerLostMidTurn(input: { wake: "user" | "system_retry" | "system_monitor"; - refusal?: "process_not_running" | "process_identity_changed" | null }) { + async function runnerLostMidTurn(input: { wake: "user" | "user_retry" | "system_retry" | "system_monitor"; + refusal?: "process_not_running" | "process_identity_changed" | null; + events?: string[]; refusalPayload?: Record }) { const f = await seed("native"); const original = await run(f.runId); - const system = input.wake !== "user"; - const reason = input.wake === "user" ? "issue_assigned" : input.wake === "system_retry" ? "transient_failure_retry" : "issue_monitor_due"; + const system = input.wake !== "user" && input.wake !== "user_retry"; + const reason = input.wake === "user" ? "issue_assigned" : input.wake === "user_retry" ? "task_retry" + : input.wake === "system_retry" ? "transient_failure_retry" : "issue_monitor_due"; + const [retryIntent] = input.wake === "user_retry" ? await db.insert(agentWakeupRequests).values({ companyId: f.companyId, + agentId: original.agentId, source: "on_demand", triggerDetail: "manual", reason: "task_retry", status: "completed", + idempotencyKey: `task-retry:${randomUUID()}`, requestedByActorType: "user", requestedByActorId: "local-board", + payload: { issueId: f.issueId } }).returning() : []; const [predecessor] = input.wake === "system_retry" ? await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: original.agentId, issueId: f.issueId, status: "failed", createdAt: new Date(Date.now() - 60 * 60_000), finishedAt: new Date(Date.now() - 50 * 60_000), @@ -48,23 +54,24 @@ describe("a remote Runner that died in the middle of a provider turn", () => { await db.update(heartbeatRuns).set({ errorCode: "native_continuation_requires_reconciliation", error: "runner_remote_process_identity_unavailable", wakeupRequestId: wake.id, responsibleUserId: "local-board", retryOfRunId: predecessor?.id ?? null, startedAt: new Date(Date.now() - 45 * 60_000), - contextSnapshot: { issueId: f.issueId, wakeReason: reason, ...(predecessor ? { retryOfRunId: predecessor.id, + contextSnapshot: { issueId: f.issueId, wakeReason: reason, ...(retryIntent ? { taskRetryRequestId: retryIntent.id, + forceFreshSession: true } : {}), ...(predecessor ? { retryOfRunId: predecessor.id, retryReason: "transient_failure", scheduledRetryAttempt: 1 } : {}) }, }).where(eq(heartbeatRuns.id, f.runId)); - for (const eventType of ["turn.submitted", "turn.started", "tool.execution.started", "turn.accepted", "tool.execution.completed"]) + for (const eventType of input.events ?? ["turn.submitted", "turn.started", "tool.execution.started", "turn.accepted", "tool.execution.completed"]) await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId: f.runId, agentId: original.agentId, eventType, stream: "system", level: "info", message: eventType, payload: { prpEvent: { eventType, turnId: TURN_ID, payload: {} } } }); if (input.refusal !== null) await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId: f.runId, agentId: original.agentId, eventType: "native.remote_runner_recovery_failed", stream: "system", level: "error", - message: "Could not verify the existing remote runner.", payload: stagingRefusal(input.refusal) }); + message: "Could not verify the existing remote runner.", payload: input.refusalPayload ?? stagingRefusal(input.refusal) }); const claim = { kind: "reattach_remote_runner" as const, runId: f.runId, leaseOwner: `boot:2:${randomUUID()}`, controllerGeneration: 2, providerAttempt: 1, restartKind: "hard" as const, recoveryRequestId: null, remote: { providerLeaseId: f.providerLeaseId, remoteCwd: f.remote } }; const service = heartbeatService(db, { environmentRuntime: f.environmentRuntime }); // Keep the agent's only slot busy so a successor stays observable as queued. await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: original.agentId, status: "running" }); - return { ...f, agentId: original.agentId, claim, service }; + return { ...f, agentId: original.agentId, claim, service, retryIntent }; } const queued = async (companyId: string) => db.select().from(heartbeatRuns) .where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.status, "queued"))); @@ -101,6 +108,28 @@ describe("a remote Runner that died in the middle of a provider turn", () => { idempotencyKey: `task-retry:restart-interrupted-turn:${f.runId}`, requestedByActorType: "user", requestedByActorId: "local-board" })]); }, 30_000); + it("continues a person's Retry whose turn start timed out and whose Runner was gone at the same-run reattachment", async () => { + const f = await runnerLostMidTurn({ wake: "user_retry", events: ["turn.submitted"], refusalPayload: { + remoteRunnerRecovery: { stage: "process_verification", reason: "process_not_running", runnerSettled: false, + runnerLifecycle: "ready", runnerDiagnostics: ["WebSocket peer closed the connection"], providerTurnIssued: false, + runnerExitEvidence: "== runnerd.process-stderr.log" } } }); + expect(await requestInterruptedNativeTurnRetry(db, f.runId)).toEqual({ requested: true, requestId: expect.any(String) }); + expect(await requestInterruptedNativeTurnRetry(db, f.runId)).toMatchObject({ requested: true }); + expect(await unknownOutcomeEvents(f.runId)).toEqual([expect.objectContaining({ payload: expect.objectContaining({ + lastTurnEvent: "turn.submitted", replayed: false }) })]); + await vi.waitFor(async () => { + await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, current => f.service.resumeRemoteStopComments(current)) ?? []); + expect(readSystemRecovery((await action(f.actionId)).evidence)?.state).toBe("complete"); + }, { timeout: 10_000, interval: 10 }); + await vi.waitFor(async () => expect(await queued(f.companyId)).toHaveLength(1), { timeout: 10_000, interval: 50 }); + const [successor] = await queued(f.companyId); + expect(successor.contextSnapshot).toMatchObject({ previousRunId: f.runId, wakeReason: "task_retry", forceFreshSession: true }); + const retries = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, f.companyId), + eq(agentWakeupRequests.reason, "task_retry"))); + expect(retries.filter(retry => retry.id !== f.retryIntent!.id && retry.idempotencyKey?.startsWith("task-retry:"))).toEqual([ + expect.objectContaining({ idempotencyKey: `task-retry:restart-interrupted-turn:${f.runId}`, requestedByActorId: "local-board" })]); + }, 30_000); + it.each([ { title: "inherits the hook's Retry intent", inherit: true }, { title: "reaches the hook's Retry run through its retry chain", inherit: false },