diff --git a/doc/plugins/PLUGIN_RUNNER_EXECUTION.md b/doc/plugins/PLUGIN_RUNNER_EXECUTION.md index fc2bc60ecd9..b22f67176ea 100644 --- a/doc/plugins/PLUGIN_RUNNER_EXECUTION.md +++ b/doc/plugins/PLUGIN_RUNNER_EXECUTION.md @@ -194,6 +194,10 @@ Project authorization is still checked for each new run. A Launcher task owns one Runner process, independently of Core heartbeat runs. Successful warm run completion retains the task and its mounts. The reuse +policy is saved with the selected native runtime, independently of workspace +copy-back records. Recovery uses that policy when no workspace-sync record exists, +so it does not stop a Runner that foreground completion retains. Failed and +cancelled runs still stop instead of staying warm. The reuse fingerprint includes the authorized project set; changing it retires the previous task before checkpoint recovery. Adoption and each warm handoff require a current, configuration-bound project-auth check against the exact retained mount scope. Core atomically @@ -205,6 +209,11 @@ allowing the new authority. The original ingress address remains unchanged. Runner accepts `connection: { mode: "external" }` on warm attachment to preserve the process owner's listener and TLS settings while rotating the run identity. +Core permits connection refresh for the saved successful run while its warm task +lease is retained, unexpired, and not retiring or stopped. PRP authentication +still applies. A superseded lease cannot acquire a connection. Retention does +not authorize preparation or submission of new work through the completed lease. + Idle retirement suspends and stops Runner, saves its exact checkpoint, writes credentials through the durable lease handoff, then completes the provider task. Core permits read-only suspension proof for the exact saved identity of a diff --git a/maps/agent-access-map/execution-environments.md b/maps/agent-access-map/execution-environments.md index 8fe4571a675..293a7023bcb 100644 --- a/maps/agent-access-map/execution-environments.md +++ b/maps/agent-access-map/execution-environments.md @@ -186,6 +186,10 @@ Current company, agent, workspace, and lease checks still gate state reads and connection acquisition. Reconnection adds no agent management permission or replacement task submission. The integration covers this with synthetic tasks; it does not establish deployed plugin recovery. +Successful warm tasks can refresh their control connection through the current +retained lease. Expired, superseded, retiring, and stopped leases remain denied. +The [focused policy tests](../../server/src/__tests__/environment-task-runtime.test.ts) +cover these checks; this is not live provider verification. This source update does not qualify live agent access, warm retention, or a provider. See the [execution contract](../../doc/plugins/PLUGIN_RUNNER_EXECUTION.md). diff --git a/server/src/__tests__/environment-task-runtime.test.ts b/server/src/__tests__/environment-task-runtime.test.ts index c0f01ae6b65..15c8dcb7b62 100644 --- a/server/src/__tests__/environment-task-runtime.test.ts +++ b/server/src/__tests__/environment-task-runtime.test.ts @@ -341,7 +341,50 @@ describe("plugin-provided Runner execution", () => { const workers = worker({ kind: "accepted", taskId: value.lease.providerLeaseId }); await expect(executeEnvironmentTask(database(value).db, workers as never, { companyId: "company", leaseId, operation: { kind: "stop" } })).resolves.toMatchObject({ taskId: value.lease.providerLeaseId }); }); - it("returns a validated connection only while the lease and run are active", async () => { + it("reconnects a successful retained warm task without granting new execution", async () => { + const value = row(); + value.lease.status = "retained"; + value.run!.status = "succeeded"; + Object.assign(value.lease, { leasePolicy: "reuse_by_environment" }); + Object.assign(value.lease.metadata, { runnerTaskDispatched: true, + runnerTaskExecutionStopped: false, runnerTaskIdentity: { ...submit.runner }, + runnerTaskLifecyclePolicy: { mode: "warm", idleTimeoutMs: 120_000 } }); + const connection = { kind: "connection", taskId: "attempt-1", endpoint: { + kind: "authenticated_websocket", websocketUrl: "wss://runner.example.test/connect", generation: "attempt-1", + secretHeaders: [{ name: "Authorization", value: "Bearer fixture" }], + } }; + const workers = worker(connection); + await expect(executeEnvironmentTask(database(value).db, workers as never, { + companyId: "company", leaseId, operation: { kind: "connection" }, + })).resolves.toEqual(connection); + expect(workers.call).toHaveBeenCalledOnce(); + for (const operation of [submit, { kind: "describe" as const }]) { + const denied = worker(); + await expect(executeEnvironmentTask(database(value).db, denied as never, { + companyId: "company", leaseId, operation, + })).rejects.toThrow("not active"); + expect(denied.call).not.toHaveBeenCalled(); + } + for (const change of ["retiring", "stopped", "undispatched", "per_turn", "expired", "superseded", "failed", "cancelled", "missing_identity", "foreign_identity", "deleted"] as const) { + const changed = structuredClone(value); + if (change === "retiring") Object.assign(changed.lease.metadata, { runnerTaskRetiring: true }); + if (change === "stopped") Object.assign(changed.lease.metadata, { runnerTaskExecutionStopped: true }); + if (change === "undispatched") Object.assign(changed.lease.metadata, { runnerTaskDispatched: false }); + if (change === "per_turn") Object.assign(changed.lease.metadata, { runnerTaskLifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } }); + if (change === "expired") changed.lease.expiresAt = new Date(0); + if (change === "superseded") changed.lease.status = "expired"; + if (change === "failed" || change === "cancelled") changed.run!.status = change; + if (change === "missing_identity") Object.assign(changed.lease.metadata, { runnerTaskIdentity: undefined }); + if (change === "foreign_identity") Object.assign(changed.lease.metadata, { runnerTaskIdentity: { ...submit.runner, runId: "other" } }); + if (change === "deleted") changed.environment = null; + const denied = worker(connection); + await expect(executeEnvironmentTask(database(changed).db, denied as never, { + companyId: "company", leaseId, operation: { kind: "connection" }, + }), change).rejects.toThrow("not active"); + expect(denied.call).not.toHaveBeenCalled(); + } + }); + it("returns a validated connection for active execution", async () => { const connection = { kind: "connection", taskId: "attempt-1", endpoint: { kind: "authenticated_websocket", websocketUrl: "wss://runner.example.test/connect", generation: "attempt-1", secretHeaders: [{ name: "Authorization", value: "Bearer fixture" }], diff --git a/server/src/__tests__/native-environment-settlement-work.test.ts b/server/src/__tests__/native-environment-settlement-work.test.ts index 16a6e147f69..7a52bc5304d 100644 --- a/server/src/__tests__/native-environment-settlement-work.test.ts +++ b/server/src/__tests__/native-environment-settlement-work.test.ts @@ -16,7 +16,7 @@ import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordin import { heartbeatService } from "../services/heartbeat.js"; import { environmentRuntimeService, type EnvironmentDriver } from "../services/environment-runtime.js"; import { environmentService } from "../services/environments.js"; -import { nativeEnvironmentSettlementWork } from "../services/native-runtime/native-environment-settlement-work.js"; +import { nativeEnvironmentSettlementDisposition, nativeEnvironmentSettlementWork } from "../services/native-runtime/native-environment-settlement-work.js"; import { notifyNativeEnvironmentSettlement } from "../services/native-environment-settlement-notifications.js"; import { finalizeNativeRun } from "../services/native-runtime/native-run-finalizer.js"; import { PaperclipControlPlanePort } from "../services/native-runtime/paperclip-control-plane-port.js"; @@ -31,6 +31,23 @@ vi.mock("../services/run-failure-report.js", async importOriginal => ({ ...await importOriginal(), reportRunFailure: vi.fn(async () => undefined), })); +describe("saved native resource policy", () => { + it.each([ + ["keep_running", true, false, "keep_running"], + ["stop_and_retain", true, false, "stop_and_retain"], + ["destroy", true, false, "destroy"], + ["keep_running", false, false, "stop_and_retain"], + ["destroy", false, false, "stop_and_retain"], + ["keep_running", true, true, "stop_and_retain"], + ["destroy", true, true, "destroy"], + ["invalid", true, false, "stop_and_retain"], + [null, true, false, "stop_and_retain"], + ] as const)("settles %s with committed=%s and pending export=%s as %s", (saved, succeeded, pending, expected) => { + expect(nativeEnvironmentSettlementDisposition({ runnerProfileJson: { nativeResourceDisposition: saved }, + resultJson: pending ? { workspaceExportRetry: { pending: true } } : null }, succeeded)).toBe(expected); + }); +}); + describe("native environment release scheduling", () => { let temporary: Awaited> | undefined; let db: Db, root: string; @@ -356,9 +373,9 @@ describe("native environment release scheduling", () => { } finally { await settings.updateExperimental({ enableWorktreeRunExecution: false }); } } }); - it("preserves keep-running retention instead of turning a completed native run into destruction", async () => { + it.each(["workspace", "runtime"])("preserves %s keep-running retention without stopping the completed Runner", async source => { const binding = await committed(); - await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceSync: { + await db.update(heartbeatRuns).set({ runnerProfileJson: source === "runtime" ? { nativeResourceDisposition: "keep_running" } : { nativeWorkspaceSync: { schema: "paperclip.native-workspace-sync/v3", state: "finalized", descriptorSha256: "a".repeat(64), baselineSha256: "b".repeat(64), finalHostSha256: "c".repeat(64), workspaceId: randomUUID(), leaseId: randomUUID(), providerLeaseId: "kept", remoteCwd: "/workspace", resourceDisposition: "keep_running", @@ -368,6 +385,18 @@ describe("native environment release scheduling", () => { await start(); await flush(); expect(release).not.toHaveBeenCalled(); expect((await environmentService(db).getLeaseById(acquired.id))?.status).toBe("retained"); await quiet(); }); + it.each(["failed", "cancelled"] as const)("stops a %s run despite its selected warm policy", async status => { + const binding = await committed(); + await db.update(heartbeatRuns).set({ status, runnerProfileJson: { nativeResourceDisposition: "keep_running" } }) + .where(eq(heartbeatRuns.id, binding.runId)); + const acquired = await environmentService(db).acquireLease({ companyId, environmentId, heartbeatRunId: binding.runId, + issueId: binding.issueId, leasePolicy: "reuse_by_environment", provider: "local", providerLeaseId: randomUUID() }); + await start(); await flush(); + expect(release).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({ resourceDisposition: "stop_and_retain" })); + expect((await environmentService(db).getLeaseById(acquired.id))?.status).toBe("released"); + expect((await read(binding.runId)).status).toBe(status); + await quiet(); + }); it.each([true, false])("delegates only periodic environment settlement, keeping manual behavior and other callbacks: %s", async delegated => { const binding = await committed(), acquired = await lease(binding); const runtimeRelease = vi.spyOn(workspaceRuntime, "releaseRuntimeServicesForRun"); diff --git a/server/src/services/environment-task-runtime.ts b/server/src/services/environment-task-runtime.ts index abc9b6b9fd8..5abe356bda3 100644 --- a/server/src/services/environment-task-runtime.ts +++ b/server/src/services/environment-task-runtime.ts @@ -73,7 +73,13 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage lease.leasePolicy === "reuse_by_environment" && ["retained", "expired"].includes(lease.status) && run?.status === "succeeded" && metadata.runnerTaskDispatched === true && (metadata.runnerTaskLifecyclePolicy as {mode?:unknown}|undefined)?.mode === "warm"; - if ((operation.kind === "read_file" || operation.kind === "describe" || operation.kind === "prepare" || operation.kind === "submit" || operation.kind === "connection" || (operation.kind === "read_state" && metadata.runnerTaskExecutionStopped !== true && !retainedCheckpointRead) || operation.kind === "archive_state" || operation.kind === "check_project_auth" || operation.kind === "copilot_metadata") && + const retainedConnection = operation.kind === "connection" && environment !== null && + lease.leasePolicy === "reuse_by_environment" && lease.status === "retained" && + run?.status === "succeeded" && savedIdentity?.runId === run.id && + metadata.runnerTaskDispatched === true && metadata.runnerTaskExecutionStopped !== true && + metadata.runnerTaskRetiring !== true && (!lease.expiresAt || lease.expiresAt.getTime() > Date.now()) && + (metadata.runnerTaskLifecyclePolicy as { mode?: unknown } | undefined)?.mode === "warm"; + if ((operation.kind === "read_file" || operation.kind === "describe" || operation.kind === "prepare" || operation.kind === "submit" || (operation.kind === "connection" && !retainedConnection) || (operation.kind === "read_state" && metadata.runnerTaskExecutionStopped !== true && !retainedCheckpointRead) || operation.kind === "archive_state" || operation.kind === "check_project_auth" || operation.kind === "copilot_metadata") && (metadata.runnerTaskRetiring === true || !environment || (!activeProbe && run?.status !== "running") || lease.status !== "active" || (lease.expiresAt && lease.expiresAt.getTime() <= Date.now()))) throw new Error("Plugin-provided Runner execution lease is not active"); if (operation.kind === "copilot_metadata" && !activeProbe) throw new Error("Copilot metadata requires a private setup lease"); const warmAttachment = !!prior && metadata.runnerTaskReused === true && diff --git a/server/src/services/heartbeat/run-state.test.ts b/server/src/services/heartbeat/run-state.test.ts index 1b9c4276a61..28faa0efa6d 100644 --- a/server/src/services/heartbeat/run-state.test.ts +++ b/server/src/services/heartbeat/run-state.test.ts @@ -313,7 +313,7 @@ describePostgres("heartbeat run state database wiring", () => { const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId: agent.id, issueId: issue.id, runtimeMode: "native", status: "failed", errorCode: "provider_transport_failed", error: "Retain saved files", nativePhase: "terminal_failure", - runnerProfileJson: { nativeWorkspaceSync: sync, savedInput: bulk }, + runnerProfileJson: { nativeWorkspaceSync: sync, nativeResourceDisposition: "keep_running", savedInput: bulk }, resultJson: { workspaceExportRetry: { requestId: "retry-source" }, stdout: bulk }, contextSnapshot: { timerClaimWasFirstHeartbeat: true, savedInput: bulk }, stdoutExcerpt: bulk, stderrExcerpt: bulk, @@ -329,7 +329,7 @@ describePostgres("heartbeat run state database wiring", () => { }); expect(await state.getRunWorkspaceSettlement(run.id)).toEqual({ id: run.id, status: run.status, error: run.error, - runnerProfileJson: { nativeWorkspaceSync: sync }, + runnerProfileJson: { nativeWorkspaceSync: sync, nativeResourceDisposition: "keep_running" }, resultJson: { workspaceExportRetry: { requestId: "retry-source" } }, contextSnapshot: { timerClaimWasFirstHeartbeat: true }, }); diff --git a/server/src/services/heartbeat/run-state.ts b/server/src/services/heartbeat/run-state.ts index d2dec92ac43..954b27d0f0a 100644 --- a/server/src/services/heartbeat/run-state.ts +++ b/server/src/services/heartbeat/run-state.ts @@ -1174,7 +1174,8 @@ export function createHeartbeatRunState(db: Db) { id: heartbeatRuns.id, status: heartbeatRuns.status, error: safeForLegacyEncoding ? heartbeatRunSqlAsciiSafeColumns.error : heartbeatRuns.error, runnerProfileJson: sql | null>`case when ${heartbeatRuns.runnerProfileJson} is null then null - else jsonb_build_object('nativeWorkspaceSync', ${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync') end`, + else jsonb_build_object('nativeWorkspaceSync', ${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync', + 'nativeResourceDisposition', ${heartbeatRuns.runnerProfileJson}->'nativeResourceDisposition') end`, resultJson: sql | null>`case when ${heartbeatRuns.resultJson} is null then null else jsonb_build_object('workspaceExportRetry', ${heartbeatRuns.resultJson}->'workspaceExportRetry') end`, contextSnapshot: sql | null>`case when ${heartbeatRuns.contextSnapshot} is null then null diff --git a/server/src/services/heartbeat/runtime-selection.test.ts b/server/src/services/heartbeat/runtime-selection.test.ts index 8badce2038a..c7708d5e078 100644 --- a/server/src/services/heartbeat/runtime-selection.test.ts +++ b/server/src/services/heartbeat/runtime-selection.test.ts @@ -354,7 +354,10 @@ describe.skipIf(!support.supported)("heartbeat runtime selection boundary", () = const selected = await selectHeartbeatRuntime(db, input); if (!selected.selected || !selected.nativeExecution) throw new Error("Expected native Claude selection"); expect(selected.nativeExecution.session.lifecyclePolicy).toEqual({ mode: "warm", idleTimeoutMs: 300_000 }); - expect((await read(run.id)).runnerProfileJson).toMatchObject({ nativeExecutionInput: { session: { lifecyclePolicy: { mode: "warm" } } } }); + expect((await read(run.id)).runnerProfileJson).toMatchObject({ + nativeResourceDisposition: "keep_running", + nativeExecutionInput: { session: { lifecyclePolicy: { mode: "warm" } } }, + }); }); it("preserves only server-owned admission and dispatch evidence when recording legacy selection", async () => { diff --git a/server/src/services/heartbeat/runtime-selection.ts b/server/src/services/heartbeat/runtime-selection.ts index fafaf38af07..1085e6a758d 100644 --- a/server/src/services/heartbeat/runtime-selection.ts +++ b/server/src/services/heartbeat/runtime-selection.ts @@ -967,6 +967,7 @@ export async function selectHeartbeatRuntime(db: Db, input: HeartbeatRuntimeSele : {}), nativeExecutionInput: lockedProfile.nativeExecutionInput ?? nativeExecution, + nativeResourceDisposition: providerResourceDispositionForRun, nativeToolContractFingerprint: nativeToolContractFingerprintForTarget( executionTarget?.kind ?? "local", diff --git a/server/src/services/native-runtime/native-environment-settlement-work.ts b/server/src/services/native-runtime/native-environment-settlement-work.ts index e957ed890c1..43c925b63af 100644 --- a/server/src/services/native-runtime/native-environment-settlement-work.ts +++ b/server/src/services/native-runtime/native-environment-settlement-work.ts @@ -11,8 +11,11 @@ import { readNativeWorkspaceSyncReference } from "./native-workspace-sync.js"; type Run = Pick; export function nativeEnvironmentSettlementDisposition(run: Run, succeeded: boolean): ProviderResourceDisposition { const reference = readNativeWorkspaceSyncReference(run.runnerProfileJson?.nativeWorkspaceSync); - return succeeded && (!run.resultJson?.workspaceExportRetry || reference?.resourceDisposition === "destroy") - ? reference?.resourceDisposition ?? "stop_and_retain" : "stop_and_retain"; + const saved = run.runnerProfileJson?.nativeResourceDisposition; + const disposition = reference?.resourceDisposition + ?? (saved === "keep_running" || saved === "stop_and_retain" || saved === "destroy" ? saved : "stop_and_retain"); + return succeeded && (!run.resultJson?.workspaceExportRetry || disposition === "destroy") + ? disposition : "stop_and_retain"; } type Entry = { id: string; companyId: string; issueId: string; dueAt: number }; @@ -28,7 +31,8 @@ export function nativeEnvironmentSettlementWork(db: Db, settle: (input: Settleme return db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, issueId: nativeRunFinalizations.issueId, phase: nativeRunFinalizations.phase, runnerProfileJson: sql`case when ${heartbeatRuns.runnerProfileJson} is null then null - else jsonb_build_object('nativeWorkspaceSync', ${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync') end`, + else jsonb_build_object('nativeWorkspaceSync', ${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync', + 'nativeResourceDisposition', ${heartbeatRuns.runnerProfileJson}->'nativeResourceDisposition') end`, resultJson: sql`case when ${heartbeatRuns.resultJson} is null then null else jsonb_build_object('workspaceExportRetry', ${heartbeatRuns.resultJson}->'workspaceExportRetry') end`, }).from(nativeRunFinalizations)