diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts index 9174f919a9c..8c17a7772ba 100644 --- a/server/src/services/explicit-native-continuation.test.ts +++ b/server/src/services/explicit-native-continuation.test.ts @@ -10,10 +10,10 @@ import { issueTreeControlService } from "./issue-tree-control.js"; import { createReleaseIssueExecution } from "../modules/wake-queue/application/use-cases.js"; import * as nativeExecutor from "./native-runtime/native-session-executor.js"; import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js"; -import { recordNativeLocalProcessStop, hasNativeLocalProcessStop, PRIOR_OWNER_FENCE_REFUSED, PROCESS_START_REQUESTED } from "./native-local-process-stop.js"; +import { recordNativeLocalProcessStop, hasNativeLocalProcessStop, PRIOR_OWNER_FENCE_REFUSED, PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED } from "./native-local-process-stop.js"; import { remoteTerminationReceipt } from "./remote-execution-termination.js"; import { createHash, randomBytes, randomUUID } from "node:crypto"; -import { mkdtemp, mkdir, writeFile, rm } from "node:fs/promises"; +import { mkdtemp, mkdir, writeFile, rm, utimes } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { and, eq, sql } from "drizzle-orm"; @@ -1273,6 +1273,87 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes }, ); + it.each(["proven", "coarse_before_request", "coarse_same_second", "state_after_request", "lineage_after_request", "previous_release", + "earlier_runner_unproven", "live_runner"] as const)( + "starts a Retry after restarts cut off a relaunch whose earlier runners were proven stopped (%s)", async kind => { + const f = await seed(); + const stateBase = await mkdtemp(join(tmpdir(), "relaunched-runner-")); + const previous = process.env.PAPERCLIP_RUNNER_STATE_DIR; + process.env.PAPERCLIP_RUNNER_STATE_DIR = stateBase; + let child: ReturnType | undefined; + try { + const nativeSessionId = randomUUID(); + const execution = { + schema: "paperclip.native-execution-input.v1", provider: { kind: "codex", model: null }, + binding: { companyId: f.companyId, issueId: f.issueId, agentId: f.agentId, runId: f.sourceRunId, executionWorkspaceId: "workspace" }, + task: { identifier: "TEST", title: "Continue", description: null, prompt: "Continue", workMode: "standard" }, + workspace: { cwd: stateBase, repoUrl: null, repoRef: null, branchName: null }, + session: { normalizedSessionId: nativeSessionId, driverKind: "codex_app_server", protocolVersion: 1, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } }, + completionContract: { id: "contract", sha256: "sha", schemaVersion: "paperclip.completion-contract.v1", + contract: { revision: "1", objective: "Continue", criteria: [{ id: "objective", requirement: "Continue" }] } }, + interactionResponses: [], credentialBindings: [], + }; + await db.update(heartbeatRuns).set({ processPid: null, processGroupId: null, nativeSessionId: null, runnerInstanceId: null, + errorCode: "adapter_failed", runnerProfileJson: { nativeExecutionInput: execution } }).where(eq(heartbeatRuns.id, f.sourceRunId)); + const event = (eventType: string, payload: Record) => appendHeartbeatRunEvent(db, { + companyId: f.companyId, runId: f.sourceRunId, agentId: f.agentId, eventType, payload }); + const host = os.hostname(), deadPid = 999999998; + await event(PROCESS_START_REQUESTED, { hostname: host, providerAttempt: 1 }); + await event(PROCESS_IDENTITY_RECORDED, { hostname: host, targetKind: "local", processPid: deadPid, processGroupId: deadPid, + processStartedAt: "2026-11-10T11:13:34.000Z" }); + if (kind !== "earlier_runner_unproven") await event("native.local_process_stopped", + { processPid: deadPid, processGroupId: deadPid, processStartedAt: "2026-11-10T11:13:34.000Z" }); + const canonical = (value: unknown): string => value && typeof value === "object" && !Array.isArray(value) + ? `{${Object.entries(value).sort(([a], [b]) => a.localeCompare(b)).map(([key, entry]) => `${JSON.stringify(key)}:${canonical(entry)}`).join(",")}}` + : JSON.stringify(value); + const root = join(stateBase, createHash("sha256").update(canonical({ + schema: "paperclip.native-session-scope.v2", companyId: f.companyId, agentId: f.agentId, + workspace: { kind: "managed", executionWorkspaceId: "workspace" }, + provider: { driverKind: "codex_app_server", identity: { kind: "codex" } }, normalizedSessionId: nativeSessionId, + })).digest("hex")); + await mkdir(join(root, "control-plane"), { recursive: true }); + await mkdir(join(root, "runner"), { recursive: true }); + await mkdir(join(root, "runner/process-lineage"), { recursive: true }); + await writeFile(join(root, "control-plane/control-plane-state.json"), JSON.stringify({ + schema: "paperclip.runner.durable.control-plane-state.v1", identity: { runId: f.sourceRunId } })); + await writeFile(join(root, "runner/runner-state.json"), JSON.stringify({ schema: "paperclip.runner.durable.state.v1" })); + await writeFile(join(root, `runner/process-lineage/writer-${deadPid}.json`), JSON.stringify({ schema: "paperclip.runner.process-lineage-writer.v1" })); + const requestedAt = new Date(Math.floor(Date.now() / 1000) * 1000 - 59_500); + const earlier = new Date(requestedAt.getTime() - 30_000), later = new Date(requestedAt.getTime() + 30_250); + const stateTimes: Partial> = { state_after_request: later, + coarse_same_second: new Date(requestedAt.getTime() - 500), coarse_before_request: new Date(requestedAt.getTime() - 3_500) }; + const stateTime = stateTimes[kind] ?? earlier; + await utimes(join(root, "runner/runner-state.json"), earlier, stateTime); + await utimes(join(root, `runner/process-lineage/writer-${deadPid}.json`), earlier, kind === "lineage_after_request" ? later : earlier); + await event(PROCESS_START_REQUESTED, { hostname: host, providerAttempt: 1, + ...(kind === "previous_release" ? {} : { requestedAt: requestedAt.toISOString() }) }); + if (kind === "live_runner") { + child = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)", "--", "--run-id", f.sourceRunId], { stdio: "ignore" }); + await once(child, "spawn"); + } + const blocked: string[] = []; + const admitted = await db.transaction(async tx => { + await tx.select().from(issues).where(eq(issues.id, f.issueId)).for("update"); + return admitExplicitNativeContinuation({ ...f, db: tx as unknown as typeof db, reason: "retry_failed_run", + failedRunId: f.sourceRunId, onBlocked: reason => blocked.push(reason) }); + }); + if (kind === "proven" || kind === "coarse_before_request") { + expect(admitted).toMatchObject({ previousRunId: f.sourceRunId, failedRunId: f.sourceRunId }); + expect(blocked).toEqual([]); + } else { + expect(admitted).toBeNull(); + expect(blocked).toEqual(["process_identity_missing"]); + } + expect(await hasNativeLocalProcessStop(db, f.companyId, f.sourceRunId)).toBe(kind === "proven" || kind === "coarse_before_request"); + } finally { + if (child) { const exited = once(child, "exit"); child.kill("SIGKILL"); await exited; } + if (previous === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; + else process.env.PAPERCLIP_RUNNER_STATE_DIR = previous; + await rm(stateBase, { recursive: true, force: true }); + } + }, + ); + it.each(["suspended", "ready", "wrong_run", "wrong_thread", "active_provider", "pending_tool", "pending_output", "missing_state", "new_launch"])( "recovers a historical run without process metadata only from exact suspended state (%s)", async kind => { const f = await seed(); diff --git a/server/src/services/native-local-process-stop.ts b/server/src/services/native-local-process-stop.ts index 3b862654348..43fcb1723c5 100644 --- a/server/src/services/native-local-process-stop.ts +++ b/server/src/services/native-local-process-stop.ts @@ -112,26 +112,30 @@ export async function hasNativeLocalProcessStop(db: Db, companyId: string, runId /** * A local runner launch can be requested and the server killed before the * runner's process identity is recorded. With no PID to check, prove on this - * host that no runner for the run exists and that it never wrote runner state - * (no provider can start before that), then record the stop like a verified - * PID. A later launch request invalidates the receipt as usual. + * host that no runner for the run exists and that this launch wrote no runner + * state (no provider can start before that), then record the stop like a + * verified PID. A later launch request invalidates the receipt as usual. */ export async function recordUnidentifiedLocalRunnerStop(db: Db, run: Pick, options: { record: boolean } = { record: true }): Promise { if (run.runtimeMode !== "native" || run.processPid || run.processGroupId) return false; - const [latest] = await db.select({ eventType: heartbeatRunEvents.eventType, payload: heartbeatRunEvents.payload }).from(heartbeatRunEvents).where(and( + const [latest, previous] = await db.select({ eventType: heartbeatRunEvents.eventType, payload: heartbeatRunEvents.payload }).from(heartbeatRunEvents).where(and( eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), isNull(heartbeatRunEvents.sourceEventId), inArray(heartbeatRunEvents.eventType, [PROCESS_START_REQUESTED, PROCESS_IDENTITY_RECORDED, LOCAL_PROCESS_STOPPED]), - )).orderBy(desc(heartbeatRunEvents.seq)).limit(1); + )).orderBy(desc(heartbeatRunEvents.seq)).limit(2); // Only the host that requested the launch can inspect its process table. if (latest?.eventType !== PROCESS_START_REQUESTED || latest.payload?.hostname !== os.hostname()) return false; const leases = await db.select({ provider: environmentLeases.provider }).from(environmentLeases) .where(and(eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id))); if (leases.some(lease => lease.provider !== "local")) return false; + const requestedAt = typeof latest.payload?.requestedAt === "string" ? Date.parse(latest.payload.requestedAt) : Number.NaN; + const launchRequestedAt = (!previous || previous.eventType === LOCAL_PROCESS_STOPPED) && Number.isFinite(requestedAt) + ? requestedAt : null; const { unidentifiedLocalRunnerProvenAbsent } = await import("./native-runtime/native-session-executor.js"); - if (!await unidentifiedLocalRunnerProvenAbsent({ runId: run.id, execution: run.runnerProfileJson?.nativeExecutionInput })) return false; + if (!await unidentifiedLocalRunnerProvenAbsent({ runId: run.id, execution: run.runnerProfileJson?.nativeExecutionInput, + launchRequestedAt })) return false; if (!options.record) return true; await appendHeartbeatRunEvent(db, { companyId: run.companyId, @@ -140,7 +144,7 @@ export async function recordUnidentifiedLocalRunnerStop(db: Db, run: Pick { +export async function unidentifiedLocalRunnerProvenAbsent(input: { + runId: string; execution: unknown; launchRequestedAt?: number | null; +}): Promise { let execution: NativeExecutionInput; try { execution = parseNativeExecutionInput(input.execution); @@ -2219,7 +2224,8 @@ export async function unidentifiedLocalRunnerProvenAbsent(input: { runId: string for (const candidate of candidates) { if (!existsSync(resolve(candidate, "control-plane", "control-plane-state.json"))) continue; try { - if (record(readRunnerdControlPlaneState(candidate).identity).runId === input.runId) return false; + if (record(readRunnerdControlPlaneState(candidate).identity).runId === input.runId && + !runnerStateUnchangedSince(candidate, input.launchRequestedAt)) return false; } catch { return false; } @@ -2227,6 +2233,22 @@ export async function unidentifiedLocalRunnerProvenAbsent(input: { runId: string return true; } +const COARSE_FILE_TIMESTAMP_MARGIN_MS = 2_000; + +function runnerStateUnchangedSince(root: string, requestedAt: number | null | undefined): boolean { + if (typeof requestedAt !== "number" || !Number.isFinite(requestedAt)) return false; + const stateFile = resolve(root, "runner", "runner-state.json"); + const lineage = resolve(root, "runner", "process-lineage"); + const paths = [stateFile, ...existsSync(lineage) ? readdirSync(lineage).map(name => resolve(lineage, name)) : []]; + return paths.every(path => { + const stats = lstatSync(path, { throwIfNoEntry: false }); + if (path === stateFile && !stats) return true; + if (!stats?.isFile()) return false; + const coarseTimestamp = stats.mtimeMs % 1000 === 0; + return stats.mtimeMs + (coarseTimestamp ? COARSE_FILE_TIMESTAMP_MARGIN_MS : 0) < requestedAt; + }); +} + /** `paperclip-runnerd` is always launched with `--run-id `. A failed or * slow process listing proves nothing, so it keeps the caller blocked. */ export async function noLocalRunnerProcessForRun(runId: string): Promise { @@ -11243,7 +11265,7 @@ async function executePaperclipNativeSessionAttempt( // Binds a later identity-less stop proof to the host that launched. // The provider attempt opens that attempt's outcome window for // the work-based budget (provider-attempt-outcome.ts). - payload: { hostname: hostname(), providerAttempt: attempt }, + payload: { hostname: hostname(), providerAttempt: attempt, requestedAt: new Date().toISOString() }, }); crashPoint("native.process_start.after_requested_before_spawn"); },