diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index d23bd27c1d4..e5533bc816c 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -2785,6 +2785,9 @@ Remote adoption tests must invoke actual external-state preparation. Inspecting transport options alone does not exercise the resumed-lease harness checks. Include a healthy warm handoff before the crash so recovery exercises the handoff's turn and item attachment IDs as well as first-run identities. +Retirement tests must exercise the preceding `turn.stop` as well as drain and +suspend after result acceptance. Configuration tests compare fresh and recovered +home-mode fingerprints while preserving rejection of real mode changes. Run it against an isolated database with `PAPERCLIP_TEST_DATABASE_URL` when the host cannot start embedded Postgres: diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index be3b5acb9ff..4cc993a5905 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2724,6 +2724,14 @@ attachment IDs. A healthy warm handoff can create new attachment IDs within the retained Runner. Replacing that run's controller must preserve those IDs; deriving them again from the business run ID would request an invalid authority rotation. +Accepted-result retirement still stops the old provider turn before draining and +suspending its Runner. The checkpoint codec classifies `turn.stop` as lifecycle +maintenance. Existing exact ownership, generation and accepted-result fences +govern this cleanup; starting provider work still requires an active owner. +Recovery also preserves the implicit isolated-home default when binding admitted +configuration, so an unchanged home does not invalidate the retained session. +Explicit home-mode changes retain their normal configuration-change checks. + For per-turn direct sandbox recovery, the executor reads the retained Runner's suspended lifecycle through the recorded sandbox lease and validates its complete identity and lack of pending delivery/cleanup. Only a lifecycle projection is diff --git a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts index b9f982401d2..36b38caba82 100644 --- a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts +++ b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts @@ -106,6 +106,7 @@ it("restores authentication and commands from opaque checkpoint records", async try { const ticket = first.issueBootstrapTicket(); first.queueCommand("session.snapshot", {}, "maintenance"); + first.queueCommand("turn.stop", {}, "retirement-maintenance"); first.queueCommand("turn.start", {}, "execution"); first.persistPreparedLaunchConfiguration(identity, "retained-configuration"); await first.flushCheckpoint(); diff --git a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts index eb67e6d9849..4b7ae503420 100644 --- a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts +++ b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts @@ -357,7 +357,7 @@ function captureCheckpoint(state: StoredCoreState, runnerVersion: string, runner records.set("header", JSON.stringify(header)); records.set("cursor", JSON.stringify({ ackedSourceSeq: state.ackedSourceSeq, replayDeliveries: state.replayDeliveries, commandCount: commands.length, eventCount: recoveryEvents.length })); - const maintenance = new Set(["runner.drain", "runner.suspend", "session.snapshot", "session.close"]); + const maintenance = new Set(["turn.stop", "runner.drain", "runner.suspend", "session.snapshot", "session.close"]); return { runnerVersion, runnerDigest, runId, records, ...checkpointChanges(records, runId), recoveryEvidence: { runnerInstanceId: state.identity.runnerInstanceId, normalizedSessionId: state.identity.normalizedSessionId, diff --git a/server/src/__tests__/heartbeat-execution-configuration.test.ts b/server/src/__tests__/heartbeat-execution-configuration.test.ts index 509a9cfa7e6..f99bcc5d672 100644 --- a/server/src/__tests__/heartbeat-execution-configuration.test.ts +++ b/server/src/__tests__/heartbeat-execution-configuration.test.ts @@ -14,6 +14,7 @@ import { registerServerAdapter, unregisterServerAdapter, type AdapterExecutionCo import { HttpError } from "../errors.js"; import { heartbeatService } from "../services/heartbeat.js"; import { bindHeartbeatRunnerHomeMode, prepareHeartbeatExecutionConfiguration, type HeartbeatExecutionConfigurationInput } from "../services/heartbeat/execution-configuration.js"; +import { buildEffectiveRunSessionConfigMetadata, resolveTaskSessionConfigFreshness } from "../services/heartbeat/workspaces.js"; import { aiRuntimeConnectionBindingSchema } from "@paperclipai/shared"; import { secretService } from "../services/secrets.js"; import { instanceSettingsService } from "../services/instance-settings.js"; @@ -26,6 +27,44 @@ import { recoveryPostgresSupport, startRecoveryPostgres } from "./helpers/recove const skillFailures = vi.hoisted(() => new Map()); describe("Runner home mode during recovery preparation", () => { + const fingerprint = (effectiveAdapterConfig: Record) => buildEffectiveRunSessionConfigMetadata({ + adapterType: "paperclip_runner", effectiveAdapterConfig, agentRuntimeConfig: {}, issueOverrides: null, + workspaceConfig: {}, environment: null, environmentEnv: null, projectEnv: null, routineEnv: null, runtimeSkills: [], + }); + it.each([ + { schema: "paperclip.native-execution-input.v5" }, + { schema: "paperclip.native-execution-input.v7", homeMode: "isolated" }, + ])("preserves the implicit isolated fingerprint through recovery and the next run: %j", async saved => { + const config = { provider: "codex", model: "fixture-model" }; + // This is the effective configuration shape built by preparation, including + // its optional property when no explicit home mode was configured. + const prepare = (input: unknown) => ({ ...config, homeMode: bindHeartbeatRunnerHomeMode(config, input).homeMode }); + const fresh = await fingerprint(prepare(null)); + const recovered = await fingerprint(prepare(saved)); + const next = await fingerprint(prepare(null)); + expect(recovered.fingerprint).toBe(fresh.fingerprint); + expect(next.fingerprint).toBe(recovered.fingerprint); + expect(resolveTaskSessionConfigFreshness({ hasTaskSession: true, configuredModel: config.model, + taskSessionParams: { __paperclipConfiguredModel: config.model, __paperclipConfigFingerprint: recovered.fingerprint, + __paperclipConfigFingerprintVersion: recovered.version, __paperclipConfigCategoryFingerprints: recovered.categoryFingerprints }, + configMetadata: next })).toMatchObject({ reset: false, changedCategories: [] }); + expect(config).not.toHaveProperty("homeMode"); + }); + + it("pins the admitted isolated mode but invalidates a genuinely preconfigured next run", async () => { + const config = { provider: "codex", model: "fixture-model", homeMode: "preconfigured" }; + const recoveredConfig = bindHeartbeatRunnerHomeMode(config, { schema: "paperclip.native-execution-input.v5" }); + expect(recoveredConfig.homeMode).toBe("isolated"); + expect(bindHeartbeatRunnerHomeMode(config, null).homeMode).toBe("preconfigured"); + const recovered = await fingerprint(recoveredConfig); + const next = await fingerprint(config); + expect(next.fingerprint).not.toBe(recovered.fingerprint); + expect(resolveTaskSessionConfigFreshness({ hasTaskSession: true, configuredModel: config.model, + taskSessionParams: { __paperclipConfiguredModel: config.model, __paperclipConfigFingerprint: recovered.fingerprint, + __paperclipConfigFingerprintVersion: recovered.version, __paperclipConfigCategoryFingerprints: recovered.categoryFingerprints }, + configMetadata: next })).toMatchObject({ reset: true, changedCategories: ["adapterConfig"] }); + }); + it.each([ ["preconfigured", { schema: "paperclip.native-execution-input.v5" }, "isolated"], ["preconfigured", { schema: "paperclip.native-execution-input.v7", homeMode: "isolated" }, "isolated"], diff --git a/server/src/services/heartbeat/execution-configuration.ts b/server/src/services/heartbeat/execution-configuration.ts index 9dd767a3811..b01d210e321 100644 --- a/server/src/services/heartbeat/execution-configuration.ts +++ b/server/src/services/heartbeat/execution-configuration.ts @@ -49,7 +49,11 @@ export function bindHeartbeatRunnerHomeMode(config: Record, per const saved = parseObject(persistedExecution); if (Object.keys(saved).length === 0) return config; // Execution schemas before v7 always used isolated homes. - return { ...config, homeMode: saved.homeMode ?? "isolated" }; + const homeMode = saved.homeMode ?? "isolated"; + // Omission already means isolated. Recovery must not add a fingerprinted + // configuration property merely to restate that unchanged default. + if (homeMode === "isolated" && config.homeMode === undefined) return config; + return { ...config, homeMode }; } export interface HeartbeatExecutionConfigurationInput { diff --git a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts index 9a0042b97f6..1a8ec6a8474 100644 --- a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts @@ -1328,10 +1328,11 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( } finally { await authority.stop(); } }); - it.each(["idle", "preparing_successor", "unclaimed_successor", "claimed_successor", "expired_preparation", "dispatching_successor", "foreign_preparation", "foreign_native_preparation", "wrong_issue", "new_work", "replaced_controller"] as const)( - "settles a completed warm checkpoint during early retirement without granting work: %s", - async scenario => { - const fixture = await seedRun(`WARM-EARLY-RETIREMENT-${scenario}`); + it.each((["idle", "preparing_successor", "unclaimed_successor", "claimed_successor", "expired_preparation", "dispatching_successor", "foreign_preparation", "foreign_native_preparation", "wrong_issue", "new_work", "replaced_controller", "unaccepted"] as const).flatMap(scenario => + (["runner.suspend", "turn.stop"] as const).map(command => ({ scenario, command }))))( + "settles a completed warm checkpoint during early retirement without granting work: $scenario/$command", + async ({ scenario, command }) => { + const fixture = await seedRun(`WARM-EARLY-RETIREMENT-${scenario}-${command}`); const binding = { companyId, issueId: fixture.issueId, runId: fixture.runId }; const identity = transportOptions(fixture, runtimeRoot).prpIdentity; const authority = new DurablePrpControlPlane({ @@ -1356,7 +1357,7 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( }); await fixture.db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }) .where(eq(heartbeatRuns.id, fixture.runId)); - await fixture.db.update(nativeRunFinalizations).set({ phase: "committed", resultId, + await fixture.db.update(nativeRunFinalizations).set({ phase: "committed", resultId: scenario === "unaccepted" ? null : resultId, leaseOwner: null, leaseExpiresAt: null, ...(scenario === "replaced_controller" ? { controllerBootId: randomUUID(), controllerGeneration: 2 } : {}), }).where(eq(nativeRunFinalizations.runId, fixture.runId)); @@ -1365,7 +1366,7 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( if (scenario !== "idle") { nextRunId = randomUUID(); const preparing = scenario !== "claimed_successor" && scenario !== "unclaimed_successor"; - const otherIssue = scenario === "wrong_issue" ? (await seedRun("WARM-OTHER-ISSUE")).issueId : fixture.issueId; + const otherIssue = scenario === "wrong_issue" ? (await seedRun(`WARM-OTHER-ISSUE-${command}`)).issueId : fixture.issueId; await fixture.db.insert(heartbeatRuns).values({ id: nextRunId, companyId, agentId, issueId: otherIssue, scopeKind: "issue", contextSnapshot: { issueId: otherIssue }, ...(preparing ? { runtimeMode: "legacy", executionStage: scenario === "dispatching_successor" ? "dispatching" : "preparing", @@ -1382,7 +1383,7 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( } await fixture.db.update(issues).set({ executionRunId: nextRunId }).where(eq(issues.id, fixture.issueId)); const before = (await loadNativeRunnerCheckpoint(fixture.db, binding))!; - authority.queueCommand(scenario === "new_work" ? "turn.start" : "runner.suspend", {}, `early-retirement-${scenario}`); + authority.queueCommand(scenario === "new_work" ? "turn.start" : command, {}, `early-retirement-${scenario}`); if (!["idle", "preparing_successor", "claimed_successor"].includes(scenario)) { await expect(authority.flushCheckpoint()).rejects.toThrow("native_runner_checkpoint_owner_lost"); expect((await loadNativeRunnerCheckpoint(fixture.db, binding))?.records).toEqual(before.records); @@ -1393,7 +1394,7 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( saved.restoreSessionDirectory(restoredRoot); const restored = JSON.parse(await readFile(resolve(restoredRoot, "control-plane", "control-plane-state.json"), "utf8")); expect(restored.identity).toEqual(identity); - expect(restored.commands).toContainEqual(expect.objectContaining({ type: "runner.suspend", status: "pending" })); + expect(restored.commands).toContainEqual(expect.objectContaining({ type: command, status: "pending" })); } const [task] = await fixture.db.select({ executionRunId: issues.executionRunId }).from(issues) .where(eq(issues.id, fixture.issueId)); @@ -1402,6 +1403,83 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( }, ); + realProcessIt("durably retires an accepted warm Runner while its same-controller successor prepares", async () => { + const fixture = await seedRun("REAL-WARM-RETIREMENT"); + const binding = { companyId, issueId: fixture.issueId, runId: fixture.runId }; + const root = resolve(runtimeRoot, fixture.runId, "real-retirement"); + const options = transportOptions(fixture, root); + const diagnostics: string[] = [], checkpointFailures: unknown[] = []; + let authority!: DurablePrpControlPlane; + await fixture.db.update(nativeRunFinalizations).set({ controllerBootId: successor.bootId, + controllerGeneration: 1, leaseOwner: "real-retirement-owner", leaseExpiresAt: new Date(Date.now() + 60_000), + }).where(eq(nativeRunFinalizations.runId, fixture.runId)); + const transport = createRunnerdCodexTransport({ ...options, + codexArgs: [...options.codexArgs, "--linger-after-turn-start"], + onDiagnostic: message => diagnostics.push(message), + controlPlaneRegistration: async core => { + authority = core; + const write = nativeRunnerCheckpointWriter(fixture.db, { ...binding, authorityRunId: fixture.runId, + bootId: successor.bootId, generation: 1, leaseOwner: "real-retirement-owner" }); + core.setCheckpointPersistence(async checkpoint => { + try { await write(checkpoint); } + catch (error) { checkpointFailures.push(error); throw error; } + }); + return registerRunnerPrpAuthority({ ...binding, authority: core }); + }, + }); + let runnerPid: number | null = null, providerPid: number | null = null; + try { + await transport.transport.request("thread/start", { cwd: root, dynamicTools: [] }); + await transport.transport.request("turn/start", { input: [{ type: "text", text: "Keep the provider alive after accepting its result." }] }); + runnerPid = transport.evidence().runnerPid; + providerPid = transport.evidence().providerPid; + expect(runnerPid).toEqual(expect.any(Number)); + expect(providerPid).toEqual(expect.any(Number)); + await waitForCondition("active real provider before accepted-result retirement", () => + authority.store.state.committedEvents.some(event => event.eventType === "turn.started")); + await authority.flushCheckpoint(); + const [run] = await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)); + const resultId = randomUUID(); + await fixture.db.insert(nativeRunResults).values({ id: resultId, ...binding, turnId: options.prpIdentity.turnId, + completionContractId: run.completionContractId!, serverFingerprint: "real-retirement-accepted", + schemaStatus: "accepted", canonicalSha256: "a".repeat(64), + resultJson: { schema: "paperclip.run_result.v1", reportedWorkDisposition: "done", summary: "Accepted before provider cleanup." }, + }); + await fixture.db.update(nativeRunFinalizations).set({ phase: "committed", resultId, leaseOwner: null, leaseExpiresAt: null }) + .where(eq(nativeRunFinalizations.runId, fixture.runId)); + await fixture.db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }) + .where(eq(heartbeatRuns.id, fixture.runId)); + const nextRunId = randomUUID(); + await fixture.db.insert(heartbeatRuns).values({ id: nextRunId, companyId, agentId, issueId: fixture.issueId, + scopeKind: "issue", status: "running", runtimeMode: "legacy", executionStage: "preparing", + invocationSource: "assignment", contextSnapshot: { issueId: fixture.issueId }, + controllerBootId: legacyControllerBootId, controllerLeaseExpiresAt: new Date(Date.now() + 60_000) }); + await fixture.db.update(issues).set({ executionRunId: nextRunId }).where(eq(issues.id, fixture.issueId)); + await expect(transport.transport.close(), diagnostics.join("\n")).resolves.toBeUndefined(); + expect(checkpointFailures).toEqual([]); + const state = JSON.parse(await readFile(resolve(root, "runner", "runner-state.json"), "utf8")); + expect(state).toMatchObject({ lifecycle: "suspended", runId: fixture.runId, + normalizedSessionId: options.prpIdentity.normalizedSessionId, runnerInstanceId: options.prpIdentity.runnerInstanceId }); + const saved = (await loadNativeRunnerCheckpoint(fixture.db, binding))!; + const commands = [...saved.records].filter(([key]) => key.startsWith("command:")).map(([, value]) => JSON.parse(value)); + for (const type of ["turn.stop", "runner.drain", "runner.suspend"]) { + expect(commands, diagnostics.join("\n")).toContainEqual(expect.objectContaining({ type, status: "completed" })); + } + expect(commands.filter(command => command.type === "turn.start")).toHaveLength(1); + const [task] = await fixture.db.select().from(issues).where(eq(issues.id, fixture.issueId)); + expect(task.executionRunId).toBe(nextRunId); + } catch (error) { + throw new Error(`${String(error)}; owned retirement diagnostics ${JSON.stringify({ + checkpointFailures: checkpointFailures.map(String), diagnostics, + commands: authority?.store.state.commands.map(({ type, status }) => ({ type, status })), + })}`); + } finally { + await transport.transport.close().catch(() => undefined); + if (providerPid && processAlive(providerPid)) process.kill(providerPid, "SIGKILL"); + await stopOwnedProcessGroup(runnerPid, root).catch(() => undefined); + } + }, 45_000); + it("applies checkpoint deltas in Postgres without rewriting authority and removes only evicted events", async () => { const fixture = await seedRun("CHECKPOINT-DELTA"); const root = resolve(runtimeRoot, fixture.runId);