mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
feat(server): preserve native Runner state across ephemeral deploys
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed
1 parent
1a39dea9aa
commit
5c2764634f
47 files changed
+1902
-107
No files matched your search
@@ -1,6 +1,6 @@
|
||||
import { afterEach, expect, it, vi } from "vitest";
|
||||
import { assertEphemeralServerConfig } from "../services/ephemeral-server-contract.js";
|
||||
import { assertEphemeralWorkspaceExecution, assertEphemeralNativeLifecycle } from "../services/remote-workspaces.js";
|
||||
import { assertEphemeralWorkspaceExecution, assertEphemeralNativeRuntime } from "../services/remote-workspaces.js";
|
||||
import { assertEphemeralAgentInstructions } from "../services/agent-instructions.js";
|
||||
|
||||
const config = { databaseUrl: "postgresql://runtime:fixture@ep-example-pooler.neon.test/tenant", storageProvider: "s3" as const };
|
||||
@@ -24,20 +24,20 @@ it("never connects the listener with migration privileges, another tenant, or th
|
||||
expect(() => assertEphemeralServerConfig(config, { ...env, PAPERCLIP_DATABASE_LISTEN_URL: url })).toThrow("runtime role");
|
||||
}
|
||||
});
|
||||
it("rejects disk-dependent execution and instructions while retaining remote warm execution", () => {
|
||||
it("requires remote workspace authority and native execution independently of provider lifecycle", () => {
|
||||
vi.stubEnv("PAPERCLIP_EPHEMERAL_SERVER", "true");
|
||||
const remote = { externalConversation: false, adapterType: "paperclip_runner", pluginTaskWorkspace: false, remoteAuthority: true };
|
||||
expect(() => assertEphemeralWorkspaceExecution(remote)).not.toThrow();
|
||||
expect(() => assertEphemeralWorkspaceExecution({ ...remote, remoteAuthority: false, pluginTaskWorkspace: true })).not.toThrow();
|
||||
for (const patch of [{ remoteAuthority: false }, { adapterType: "process" }]) {
|
||||
expect(() => assertEphemeralWorkspaceExecution({ ...remote, ...patch })).toThrow("warm Paperclip Runner");
|
||||
expect(() => assertEphemeralWorkspaceExecution({ ...remote, ...patch })).toThrow("native Paperclip Runner");
|
||||
}
|
||||
expect(() => assertEphemeralNativeLifecycle("warm")).not.toThrow();
|
||||
expect(() => assertEphemeralNativeLifecycle("per_turn")).toThrow("warm Paperclip Runner");
|
||||
expect(() => assertEphemeralNativeLifecycle(undefined)).toThrow("warm Paperclip Runner");
|
||||
expect(() => assertEphemeralNativeRuntime(true)).not.toThrow();
|
||||
expect(() => assertEphemeralNativeRuntime(false)).toThrow("native Paperclip Runner");
|
||||
const agent = { id: "agent", companyId: "company", name: "Agent", adapterConfig: {} };
|
||||
expect(() => assertEphemeralAgentInstructions(agent)).not.toThrow();
|
||||
expect(() => assertEphemeralAgentInstructions({ ...agent, adapterConfig: { instructionsBundleMode: "external", instructionsRootPath: "/old-volume" } })).toThrow("managed agent instructions");
|
||||
vi.stubEnv("PAPERCLIP_EPHEMERAL_SERVER", "false");
|
||||
expect(() => assertEphemeralWorkspaceExecution({ ...remote, remoteAuthority: false })).not.toThrow();
|
||||
expect(() => assertEphemeralNativeRuntime(false)).not.toThrow();
|
||||
});
|
||||
@@ -9,6 +9,8 @@ import { skillPackageMaintenance } from "../services/skill-package-maintenance.j
|
||||
import { hasUnsettledLocalInstructionCopies } from "../services/agent-instruction-working-copies.js";
|
||||
import { providerTraceStore } from "../services/provider-trace-store.js";
|
||||
import { migrateWorkspaceTrees } from "../services/remote-workspace-migration-batch.js";
|
||||
import { migrateNativeHarnessBackups } from "../services/native-runtime/native-harness-backup-migration.js";
|
||||
import { migrateNativeControllers } from "../services/native-runtime/native-controller-migration.js";
|
||||
|
||||
/** Runs only in the maintenance process after Cloud has confirmed all legacy
|
||||
* deployments stopped. Each owner retains its own migration/publication rules;
|
||||
@@ -25,12 +27,21 @@ export async function migrateEphemeralServerData(db: Db, input: {
|
||||
const before = await readEphemeralServerDataEligibility(db);
|
||||
const unsupported = before.blockers.filter(blocker => ![
|
||||
"agent_files_require_migration", "run_logs_require_migration", "provider_traces_require_migration", "skills_require_migration", "workspaces_require_migration",
|
||||
"native_harness_backups_require_migration",
|
||||
"native_controllers_require_migration",
|
||||
].includes(blocker));
|
||||
if (unsupported.length) throw new Error(unsupported[0]);
|
||||
|
||||
phase("workspaces");
|
||||
const storage = getStorageService();
|
||||
if (storage.provider !== "s3") throw new Error("shared_object_storage_required");
|
||||
phase("native_controllers");
|
||||
const controllers = await migrateNativeControllers(db, { writersStopped: true, signal });
|
||||
counts.nativeControllers = controllers.imported;
|
||||
counts.nativeControllersUnchanged = controllers.unchanged;
|
||||
counts.nativeControllerSourcesAlreadyAbsent = controllers.sourceAlreadyAbsent;
|
||||
phase("native_harness_backups");
|
||||
counts.nativeHarnessBackups = await migrateNativeHarnessBackups(db, { storage, writersStopped: true, signal });
|
||||
phase("workspaces");
|
||||
counts.workspaces = await migrateWorkspaceTrees(db, { storage, signal, phase });
|
||||
|
||||
phase("agent_files");
|
||||
|
||||
@@ -56,7 +56,7 @@ import {
|
||||
import { environmentService } from "./environments.js";
|
||||
import { instanceSettingsService } from "./instance-settings.js";
|
||||
import { hasSandboxSnapshotQualification, qualifiedSandboxSnapshotId, usesManagedCloudSandboxImage, type SandboxTemplateIdentity } from "./sandbox-snapshot-qualification.js";
|
||||
import { verifyNativeHarnessBackupStamp } from "./native-runtime/native-harness-backup-stamp.js";
|
||||
import { verifyRecoverableNativeHarnessBackup } from "./native-runtime/native-harness-backup-storage.js";
|
||||
import {
|
||||
collectEnvironmentSecretRefs,
|
||||
parseEnvironmentDriverConfig,
|
||||
@@ -1847,10 +1847,8 @@ function createSandboxEnvironmentDriver(
|
||||
if (!providerLease) {
|
||||
if (
|
||||
input.adapterType === "paperclip_runner" &&
|
||||
!verifyNativeHarnessBackupStamp(
|
||||
reusableLease.metadata?.nativeHarnessBackup,
|
||||
reusableLease.providerLeaseId,
|
||||
)
|
||||
!await verifyRecoverableNativeHarnessBackup({ db, companyId: reusableLease.companyId,
|
||||
stamp: reusableLease.metadata?.nativeHarnessBackup, providerLeaseId: reusableLease.providerLeaseId })
|
||||
) {
|
||||
throw new RunnerHarnessBackupUnavailableError(
|
||||
reusableLease.providerLeaseId,
|
||||
@@ -3986,10 +3984,8 @@ export function environmentRuntimeService(
|
||||
"paperclip_runner" &&
|
||||
leaseSnapshot.metadata?.sandboxLeaseAcquisition &&
|
||||
(!leaseSnapshot.providerLeaseId ||
|
||||
!verifyNativeHarnessBackupStamp(
|
||||
leaseSnapshot.metadata.nativeHarnessBackup,
|
||||
leaseSnapshot.providerLeaseId,
|
||||
))
|
||||
!await verifyRecoverableNativeHarnessBackup({ db, companyId: leaseSnapshot.companyId,
|
||||
stamp: leaseSnapshot.metadata.nativeHarnessBackup, providerLeaseId: leaseSnapshot.providerLeaseId }))
|
||||
) {
|
||||
throw new RunnerHarnessBackupUnavailableError(
|
||||
leaseSnapshot.providerLeaseId ?? leaseSnapshot.id,
|
||||
|
||||
@@ -4,6 +4,8 @@ import type { Config } from "../config.js";
|
||||
import { isValidLocalEncryptedMasterKey } from "../secrets/local-encrypted-provider.js";
|
||||
import { agentInstructionsBundleMode } from "./agent-instructions.js";
|
||||
import { skillInstallationMaintenance } from "./skill-installation-maintenance.js";
|
||||
import { hasLocalNativeHarnessObligations } from "./native-runtime/native-harness-backup-migration.js";
|
||||
import { hasLocalNativeControllerObligations } from "./native-runtime/native-controller-migration.js";
|
||||
|
||||
export const ephemeralServerEnabled = () => process.env.PAPERCLIP_EPHEMERAL_SERVER === "true";
|
||||
|
||||
@@ -64,6 +66,8 @@ export async function readEphemeralServerDataEligibility(db: Db, options: { purp
|
||||
isNull(providerTraceRecords.deletedAt), sql`${providerTraceRecords.traceRef} not like 's3-segments-v1:%'`,
|
||||
)).limit(1);
|
||||
if (localTrace) blockers.push("provider_traces_require_migration");
|
||||
if (await hasLocalNativeHarnessObligations(db)) blockers.push("native_harness_backups_require_migration");
|
||||
if (await hasLocalNativeControllerObligations(db)) blockers.push("native_controllers_require_migration");
|
||||
const skills = skillInstallationMaintenance(db);
|
||||
let after: string | undefined;
|
||||
for (;;) {
|
||||
|
||||
@@ -432,14 +432,14 @@ describe.skipIf(!support.supported)("heartbeat runtime selection boundary", () =
|
||||
const { input, run } = await fixture("process");
|
||||
vi.stubEnv("PAPERCLIP_EPHEMERAL_SERVER", "true");
|
||||
expect(input.nativeRuntimeResolution.kind).toBe("legacy");
|
||||
await expect(selectHeartbeatRuntime(db, input)).rejects.toThrow("warm Paperclip Runner");
|
||||
await expect(selectHeartbeatRuntime(db, input)).rejects.toThrow("native Paperclip Runner");
|
||||
expect((await read(run.id)).runtimeMode).toBe("legacy");
|
||||
expect(await coordinators(run.id)).toHaveLength(0);
|
||||
expect(input.onNativeLifecycleSelected).not.toHaveBeenCalled();
|
||||
expect(prepareNativeWorkspaceSync).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it.each(["warm", "per_turn"] as const)("checks resolved %s policy for ephemeral admission", async mode => {
|
||||
it.each(["warm", "per_turn"] as const)("preserves resolved %s policy on an ephemeral server", async mode => {
|
||||
const { input, run } = await fixture();
|
||||
vi.stubEnv("PAPERCLIP_EPHEMERAL_SERVER", "true");
|
||||
// Deliberately disagree with the resolved provider policy in both cases.
|
||||
@@ -447,14 +447,41 @@ describe.skipIf(!support.supported)("heartbeat runtime selection boundary", () =
|
||||
input.workspace.executionTarget = { ...sandbox, reusableLeaseConfigured: true,
|
||||
effectiveCapabilities: { reusableLeases: true } as never,
|
||||
runnerLifecyclePolicy: mode === "warm" ? { mode, idleTimeoutMs: 300_000 } : { mode, idleTimeoutMs: null } };
|
||||
if (mode === "per_turn") {
|
||||
await expect(selectHeartbeatRuntime(db, input)).rejects.toThrow("warm Paperclip Runner");
|
||||
expect(await coordinators(run.id)).toHaveLength(0);
|
||||
} else {
|
||||
const selected = await selectHeartbeatRuntime(db, input);
|
||||
expect(selected.selected && selected.nativeExecution?.session.lifecyclePolicy.mode).toBe("warm");
|
||||
expect((await read(run.id)).runnerProfileJson).toMatchObject({ nativeExecutionInput: { session: { lifecyclePolicy: { mode: "warm" } } } });
|
||||
}
|
||||
const selected = await selectHeartbeatRuntime(db, input);
|
||||
expect(selected.selected && selected.nativeExecution?.session.lifecyclePolicy.mode).toBe(mode);
|
||||
expect((await read(run.id)).runnerProfileJson).toMatchObject({ nativeExecutionInput: { session: { lifecyclePolicy: { mode } } } });
|
||||
expect(await coordinators(run.id)).toHaveLength(1);
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ provider: "acpx", acpxAgent: "claude", model: "claude-opus-5-5", credentialProvider: "anthropic" as const, method: "subscription" as const },
|
||||
{ provider: "acpx", acpxAgent: "grok", model: "grok-4.7", credentialProvider: "xai" as const, method: "subscription" as const },
|
||||
{ provider: "opencode", model: "openrouter/deepseek/deepseek-v4-flash-0731", credentialProvider: "openrouter" as const, method: "api_key" as const },
|
||||
])("admits managed $provider/$acpxAgent on an ephemeral server without changing credential lifecycle", async profile => {
|
||||
vi.stubEnv("PAPERCLIP_EPHEMERAL_SERVER", "true");
|
||||
const { input, run, issue } = await fixture();
|
||||
const { credentialProvider, method, ...providerConfig } = profile;
|
||||
const adapterConfig = { ...providerConfig, lifecycleMode: "warm" };
|
||||
input.agent = { ...input.agent, adapterConfig };
|
||||
await db.update(agents).set({ adapterConfig }).where(eq(agents.id, input.agent.id));
|
||||
input.workspace.executionTarget = { ...sandbox, reusableLeaseConfigured: true,
|
||||
effectiveCapabilities: { reusableLeases: true } as never, runnerLifecyclePolicy: { mode: "warm", idleTimeoutMs: 300_000 } };
|
||||
input.nativeRuntimeResolution = resolveHeartbeatNativeRuntimeMode({ persisted: run, enabled: true, runtimeConfig: {}, adapterConfig,
|
||||
agent: input.agent, issue, target: input.workspace.executionTarget, workspaceId: null });
|
||||
expect(input.nativeRuntimeResolution.kind).toBe("native");
|
||||
const attribution = { connectionId: randomUUID(), grantId: randomUUID(), provider: credentialProvider, method,
|
||||
mode: "responsible_user" as const, responsibleUserId: "task-owner" };
|
||||
input.config.managedAiRuntime = { sessionIdentity: "account:epoch-1", identity: "account:generation-1",
|
||||
config: { env: {}, managedAiConnection: { ...attribution, identity: "account:generation-1", sessionIdentity: "account:epoch-1" } },
|
||||
attribution, accountName: "Saved account", accountOwnerUserId: "task-owner", home: instructionRoot, cleanup: async () => {} };
|
||||
const selected = await selectHeartbeatRuntime(db, input);
|
||||
if (!selected.selected || !selected.nativeExecution) throw new Error("Expected native selection");
|
||||
expect(selected.nativeExecution.provider.kind).toBe(profile.provider);
|
||||
expect(selected.nativeExecution.session.lifecyclePolicy).toEqual({ mode: "per_turn", idleTimeoutMs: null });
|
||||
expect((await read(run.id)).runnerProfileJson).toMatchObject({
|
||||
nativeExecutionInput: { session: { lifecyclePolicy: { mode: "per_turn" } } },
|
||||
});
|
||||
expect(await coordinators(run.id)).toHaveLength(1);
|
||||
});
|
||||
|
||||
it.each([
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import { resolveProviderApiPlacement } from "../native-runtime/provider-api-placement.js";
|
||||
import { readClaimedNativeStopRecovery } from "../native-runtime/native-cancellation-request.js";
|
||||
import { museRunnerBroker } from "../muse-runner-broker.js";
|
||||
import { assertEphemeralNativeLifecycle, readRemoteWorkspaceAuthority } from "../remote-workspaces.js";
|
||||
import { assertEphemeralNativeRuntime, readRemoteWorkspaceAuthority } from "../remote-workspaces.js";
|
||||
import { measureRunTransaction } from "../execution-transaction-timing.js";
|
||||
import { measureRunPhase } from "../execution-preparation-timing.js";
|
||||
import {
|
||||
@@ -269,7 +269,7 @@ export async function selectHeartbeatRuntime(db: Db, input: HeartbeatRuntimeSele
|
||||
const { runtimeConfig, runtimeSkillEntries, runScopedMentionedSkillKeys, instructionCopy, agentIdentity, issueAssigneeOverrides, resolvedConfig, managedAiRuntime } = input.config;
|
||||
const runOptions = input.recovery;
|
||||
const { providerTraceRequested, providerTraceCapture } = input.trace;
|
||||
if (nativeRuntimeResolution.kind !== "native" && !isDotRun) assertEphemeralNativeLifecycle(undefined);
|
||||
if (!isDotRun) assertEphemeralNativeRuntime(nativeRuntimeResolution.kind === "native");
|
||||
let nativeExecution: NativeExecutionInput | null = null;
|
||||
let getNativeFreshSessionHandoff: (() => Promise<string | null>) | undefined;
|
||||
let nativeRunnerInstanceId: string | null = null;
|
||||
@@ -522,9 +522,6 @@ export async function selectHeartbeatRuntime(db: Db, input: HeartbeatRuntimeSele
|
||||
(["openai_dot_mcp", "grok_bot_mcp", "muse_external", "cursor_cloud_api"].includes(nativeRuntimeResolution.profile.backend) || managedAiRuntime && !supportsManagedWarmSession
|
||||
? { mode: "per_turn" as const, idleTimeoutMs: null }
|
||||
: environmentLifecyclePolicy ?? agentLifecyclePolicy);
|
||||
if (!["openai_dot_mcp", "grok_bot_mcp", "muse_external", "cursor_cloud_api"].includes(nativeRuntimeResolution.profile.backend)) {
|
||||
assertEphemeralNativeLifecycle(effectiveLifecyclePolicy.mode);
|
||||
}
|
||||
if (
|
||||
effectiveLifecyclePolicy.mode === "warm" &&
|
||||
executionTarget?.kind === "remote" &&
|
||||
|
||||
@@ -88,13 +88,14 @@ postgresTests("upgrade readiness persisted execution evidence", () => {
|
||||
finally { await core.stop(); await rm(root, { recursive: true, force: true }); }
|
||||
expect(await readUpgradeReadiness(db, input)).toMatchObject({ ready: true, recoverableRuns: 1 });
|
||||
const cipher = createLocalEncryptedCipher("write");
|
||||
// Real staging headers contain retained launch metadata around 113 KiB.
|
||||
const representativeHeader = JSON.stringify({ ...JSON.parse(header), padding: "x".repeat(100_000) });
|
||||
const parsedHeader = JSON.parse(header);
|
||||
const representativeHeader = JSON.stringify({ ...parsedHeader, state: { ...parsedHeader.state,
|
||||
runAttachTemplate: { provider: { instructions: "x".repeat(300_000) } } } });
|
||||
await db.update(nativeRunnerCheckpoints).set({ material: cipher.encrypt(JSON.stringify({ ...binding, part: "header", contents: representativeHeader })) })
|
||||
.where(eq(nativeRunnerCheckpoints.runId, runId));
|
||||
expect((await readUpgradeReadiness(db, input)).ready).toBe(true);
|
||||
for (const contents of ["{", "{}", JSON.stringify({ ...JSON.parse(header), schema: "paperclip.runner.controller-checkpoint.v1" }),
|
||||
JSON.stringify({ ...JSON.parse(header), padding: "x".repeat(400_000) }),
|
||||
JSON.stringify({ ...JSON.parse(header), padding: "x".repeat(1_100_000) }),
|
||||
JSON.stringify({ ...JSON.parse(header), state: { ...JSON.parse(header).state, identity: { ...JSON.parse(header).state.identity, runId: randomUUID() } } })]) {
|
||||
await db.update(nativeRunnerCheckpoints).set({ material: cipher.encrypt(JSON.stringify({ ...binding, part: "header", contents })) })
|
||||
.where(eq(nativeRunnerCheckpoints.runId, runId));
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { sql } from "drizzle-orm";
|
||||
import { remoteWorkspaceRunReferenceSchema } from "@paperclipai/shared";
|
||||
import { inspectNativeRunnerUpgradeHeader } from "../native-runtime/native-runner-checkpoint.js";
|
||||
import { inspectNativeRunnerUpgradeHeader, NATIVE_CHECKPOINT_HEADER_INSPECTION_MAX_BYTES,
|
||||
NATIVE_CHECKPOINT_INSPECTION_TOTAL_MAX_BYTES } from "../native-runtime/native-runner-checkpoint.js";
|
||||
import { readNativeWorkspaceSyncReference } from "../native-runtime/native-workspace-sync.js";
|
||||
import { REMOTE_RUNNER_RECONNECT_GRACE_MS } from "../native-runtime/native-runner-recovery-contract.js";
|
||||
import { currentNativeControllerIdentity } from "../native-runtime/native-restart-recovery.js";
|
||||
@@ -30,7 +31,8 @@ export async function readUpgradeReadiness(db: Db, input: {
|
||||
r.runner_profile_json->'nativeWorkspaceSync' as workspace,
|
||||
r.runner_profile_json->'remoteWorkspace' as "remoteWorkspace",
|
||||
r.runner_profile_json ? 'remoteWorkspace' as "hasRemoteWorkspace", p.id as "pluginLeaseId",
|
||||
case when pg_column_size(c.material) <= 262144 and sum(pg_column_size(c.material)) over () <= 8388608 then c.material else null end as material, (
|
||||
case when pg_column_size(c.material) <= ${NATIVE_CHECKPOINT_HEADER_INSPECTION_MAX_BYTES}
|
||||
and sum(pg_column_size(c.material)) over () <= ${NATIVE_CHECKPOINT_INSPECTION_TOTAL_MAX_BYTES} then c.material else null end as material, (
|
||||
r.runtime_mode = 'native' and a.adapter_type = 'paperclip_runner'
|
||||
and r.runner_instance_id is not null and r.native_session_id is not null
|
||||
and r.process_pid is not null and r.process_started_at is not null
|
||||
|
||||
@@ -444,7 +444,7 @@ describe.skipIf(!support.supported && !externalDatabaseUrl)("heartbeat workspace
|
||||
const resolve = vi.spyOn(input.services, "resolveWorkspaceForRun").mockRejectedValue(new Error("local folder retained"));
|
||||
process.env.PAPERCLIP_EPHEMERAL_SERVER = String(ephemeral);
|
||||
await expect(prepareHeartbeatWorkspace(db, input)).rejects.toThrow(ephemeral
|
||||
? "Volume-free servers require warm Paperclip Runner execution with remote workspace storage." : "local folder retained");
|
||||
? "Volume-free servers require native Paperclip Runner execution with remote workspace storage." : "local folder retained");
|
||||
if (ephemeral) expect(resolve).not.toHaveBeenCalled();
|
||||
else expect(resolve).toHaveBeenCalledOnce();
|
||||
expect(await db.select().from(executionWorkspaces)).toHaveLength(0);
|
||||
|
||||
@@ -0,0 +1,106 @@
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { mkdir, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { afterEach, expect, it, vi } from "vitest";
|
||||
import { agentTaskSessions, heartbeatRuns, issues, nativeRunFinalizations, nativeRunnerCheckpoints } from "@paperclipai/db";
|
||||
import { DurablePrpControlPlane, parseNativeExecutionInput } from "../../vendor/paperclip-runner/index.js";
|
||||
import { describeEmbeddedPostgres, nativeRestartFixture } from "../../__tests__/helpers/native-restart.js";
|
||||
import { nativeToolContractFingerprintForTarget } from "./native-session-resume.js";
|
||||
import { nativeControllerMigrationDescriptor } from "./native-session-executor.js";
|
||||
import { hasLocalNativeControllerObligations, migrateNativeControllers } from "./native-controller-migration.js";
|
||||
import { resolveNativeControllerRecovery } from "./native-controller-recovery.js";
|
||||
import { resolveRemoteProviderPackAuthority } from "./remote-provider-pack-authority.js";
|
||||
import { canonicalProviderPackJson } from "./remote-provider-pack-verification.js";
|
||||
import { nativeSha256 } from "./canonical.js";
|
||||
|
||||
describeEmbeddedPostgres("offline native controller migration", () => {
|
||||
const environment = nativeRestartFixture();
|
||||
afterEach(() => vi.unstubAllEnvs());
|
||||
it("imports an exact settled authority once, preserves queued work, and resumes from an empty home", async () => {
|
||||
vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "73".repeat(32));
|
||||
const f = await environment.seedRun(randomUUID());
|
||||
const { companyId, agentId, paperclipHome, runtimeRoot } = environment.current;
|
||||
const binding = { companyId, agentId, issueId: f.issueId, runId: f.runId, executionWorkspaceId: f.native.environmentLeaseId };
|
||||
const execution = parseNativeExecutionInput({
|
||||
schema: "paperclip.native-execution-input.v1", provider: { kind: "codex", model: null }, binding,
|
||||
task: { identifier: "MIG-1", title: "Retained work", description: null, prompt: "Continue the work.", workMode: "standard" },
|
||||
workspace: { cwd: "/workspace", repoUrl: null, repoRef: null, branchName: null },
|
||||
session: { normalizedSessionId: f.native.normalizedSessionId, driverKind: "codex_app_server", protocolVersion: 1,
|
||||
lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } },
|
||||
completionContract: { id: randomUUID(), sha256: "fixture", schemaVersion: "paperclip.completion-contract.v1",
|
||||
contract: { revision: "1", objective: "Continue", criteria: [{ id: "objective", requirement: "Save work." }] } },
|
||||
credentialBindings: [], interactionResponses: [],
|
||||
});
|
||||
const fingerprint = nativeToolContractFingerprintForTarget("remote", "codex");
|
||||
const paths = nativeControllerMigrationDescriptor(execution, fingerprint);
|
||||
const identity = { runId: f.runId, runnerInstanceId: f.native.runnerInstanceId, normalizedSessionId: f.native.normalizedSessionId,
|
||||
environmentLeaseId: f.native.environmentLeaseId, turnId: f.native.turnId, itemId: f.native.itemId };
|
||||
const core = new DurablePrpControlPlane({ stateDirectory: join(paths.root, "control-plane"), identity,
|
||||
expectedRunnerVersion: "migration-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` });
|
||||
core.issueBootstrapTicket(); core.queueCommand("session.snapshot", {}, "retained-command"); await core.stop();
|
||||
const file = join(paths.root, "control-plane/control-plane-state.json");
|
||||
const state = JSON.parse(await readFile(file, "utf8"));
|
||||
state.tickets = {};
|
||||
const source = JSON.stringify(state); await writeFile(file, source, { mode: 0o600 });
|
||||
const packBinding = { companyId, agentId, workspaceId: binding.executionWorkspaceId,
|
||||
sessionId: f.native.normalizedSessionId, runnerInstanceId: f.native.runnerInstanceId,
|
||||
environmentLeaseId: f.native.environmentLeaseId, remoteCwd: "/remote/workspace", providerDigest: nativeSha256(execution.provider) };
|
||||
const payload = {
|
||||
runnerSourceRevision: "a".repeat(40), target: { platform: "linux", architecture: "x64" },
|
||||
pins: { opencode: "1.18.34" }, artifacts: {
|
||||
nodeCommand: { path: "node", sha256: `sha256:${"1".repeat(64)}` },
|
||||
opencodeCommand: { path: "opencode", sha256: `sha256:${"2".repeat(64)}` },
|
||||
},
|
||||
};
|
||||
const pack = resolveRemoteProviderPackAuthority({ stateDirectory: paths.root, reattach: false, binding: packBinding,
|
||||
current: () => ({ schema: "paperclip-runner/remote-provider-pack/v1", payload,
|
||||
digest: `sha256:${createHash("sha256").update(canonicalProviderPackJson(payload)).digest("hex")}` }),
|
||||
runnerArtifact: () => ({ version: "migration-fixture", digest: `sha256:${"a".repeat(64)}` }) });
|
||||
pack.verified();
|
||||
const snapshot = { backendKind: "mock", sessionId: f.native.normalizedSessionId,
|
||||
identity: { companyId, agentId, issueId: f.issueId, runId: f.runId, sessionId: f.native.normalizedSessionId } };
|
||||
const profile = { ...f.preparedProfile, nativeExecutionInput: execution, nativeToolContractFingerprint: fingerprint, sessionCheckpoint: snapshot,
|
||||
nativeWorkspaceSync: { schema: "paperclip.native-workspace-sync/v1", state: "prepared", descriptorSha256: "a".repeat(64),
|
||||
baselineSha256: "b".repeat(64), finalHostSha256: null, workspaceId: randomUUID(), leaseId: f.native.environmentLeaseId,
|
||||
providerLeaseId: randomUUID(), remoteCwd: "/remote/workspace", resourceDisposition: "destroy" } };
|
||||
await f.db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date(), runnerProfileJson: profile }).where(eq(heartbeatRuns.id, f.runId));
|
||||
await f.db.update(nativeRunFinalizations).set({ phase: "terminal_failure", leaseOwner: null }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
const queuedId = randomUUID();
|
||||
await f.db.insert(heartbeatRuns).values({ id: queuedId, companyId, agentId, issueId: f.issueId, status: "queued" });
|
||||
await f.db.update(issues).set({ executionRunId: queuedId }).where(eq(issues.id, f.issueId));
|
||||
const [session] = await f.db.insert(agentTaskSessions).values({ companyId, agentId, adapterType: "paperclip_runner",
|
||||
taskKey: f.issueId, lastRunId: f.runId }).returning();
|
||||
expect(await hasLocalNativeControllerObligations(f.db)).toBe(true);
|
||||
await f.db.update(heartbeatRuns).set({ runnerProfileJson: { ...profile,
|
||||
nativeWorkspaceSync: { ...profile.nativeWorkspaceSync, leaseId: randomUUID() } } }).where(eq(heartbeatRuns.id, f.runId));
|
||||
await expect(migrateNativeControllers(f.db, { writersStopped: true, signal: AbortSignal.timeout(30_000) }))
|
||||
.rejects.toThrow("native_controller_migration_artifact_binding");
|
||||
await f.db.update(heartbeatRuns).set({ runnerProfileJson: profile }).where(eq(heartbeatRuns.id, f.runId));
|
||||
await f.db.update(nativeRunFinalizations).set({ leaseOwner: "unsettled-owner" }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
await expect(migrateNativeControllers(f.db, { writersStopped: true, signal: AbortSignal.timeout(30_000) })).rejects.toThrow("normal_ownership_recovery_required");
|
||||
expect(await f.db.select().from(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId))).toHaveLength(0);
|
||||
await f.db.update(nativeRunFinalizations).set({ leaseOwner: "expired-owner", leaseExpiresAt: new Date(0),
|
||||
phase: "retryable_failure", recoveryState: "resuming_session" }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
const [preservedOwner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
expect(await migrateNativeControllers(f.db, { writersStopped: true, signal: AbortSignal.timeout(30_000) }))
|
||||
.toMatchObject({ imported: 1, sourceAlreadyAbsent: 0 });
|
||||
expect(await migrateNativeControllers(f.db, { writersStopped: true, signal: AbortSignal.timeout(30_000) })).toMatchObject({ unchanged: 1, imported: 0 });
|
||||
expect(await hasLocalNativeControllerObligations(f.db)).toBe(false);
|
||||
expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toEqual(preservedOwner);
|
||||
expect(await readFile(file, "utf8")).toBe(source);
|
||||
expect((await f.db.select().from(agentTaskSessions).where(eq(agentTaskSessions.id, session.id)))[0].lastRunId).toBe(f.runId);
|
||||
expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, queuedId)))[0].status).toBe("queued");
|
||||
await rm(paperclipHome, { recursive: true, force: true });
|
||||
await f.db.update(heartbeatRuns).set({ nativeIssueId: f.issueId, nativeSessionId: f.native.normalizedSessionId,
|
||||
status: "running", runtimeMode: "native" }).where(eq(heartbeatRuns.id, queuedId));
|
||||
const recovered = await resolveNativeControllerRecovery(f.db, { companyId, agentId, issueId: f.issueId,
|
||||
normalizedSessionId: f.native.normalizedSessionId, currentRunId: queuedId });
|
||||
expect(recovered.kind).toBe("recoverable");
|
||||
if (recovered.kind !== "recoverable") throw new Error(JSON.stringify(recovered));
|
||||
const replacement = join(runtimeRoot, "replacement"); await mkdir(replacement, { mode: 0o700 });
|
||||
recovered.checkpoint.restoreSessionDirectory(replacement);
|
||||
expect(JSON.parse(await readFile(join(replacement, "control-plane/control-plane-state.json"), "utf8")).identity).toEqual(identity);
|
||||
expect(recovered.snapshot).toEqual(snapshot);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,142 @@
|
||||
import { constants } from "node:fs";
|
||||
import { lstat, open, readdir } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { heartbeatRuns, type Db } from "@paperclipai/db";
|
||||
import { parseNativeExecutionInput } from "../../vendor/paperclip-runner/index.js";
|
||||
import { offlineMigrationExecutionWhere } from "../offline-migration-execution.js";
|
||||
import { findNativeSessionResumeRun } from "./native-session-resume.js";
|
||||
import { nativeControllerMigrationDescriptor } from "./native-session-executor.js";
|
||||
import { hasNativeRunnerCheckpoint, loadNativeRunnerCheckpoint } from "./native-runner-checkpoint.js";
|
||||
import { importNativeControllerRecovery, readNativeControllerRecovery } from "./native-controller-recovery.js";
|
||||
import { nativeHarnessStateBase } from "./native-harness-backup-stamp.js";
|
||||
import { providerPackWorkspaceId, resolveRemoteProviderPackAuthority } from "./remote-provider-pack-authority.js";
|
||||
import { nativeSha256 } from "./canonical.js";
|
||||
import { readNativeWorkspaceSyncReference } from "./native-workspace-sync.js";
|
||||
|
||||
type Selected = { id: string; companyId: string; agentId: string; issueId: string; sessionId: string; createdAt: Date };
|
||||
|
||||
async function* retainedSources(db: Db, signal?: AbortSignal) {
|
||||
let after: string | undefined;
|
||||
for (;;) {
|
||||
signal?.throwIfAborted();
|
||||
const rows = await db.execute<Selected>(sql`select s.id, s.company_id as "companyId", s.agent_id as "agentId",
|
||||
r.native_issue_id as "issueId", r.native_session_id as "sessionId", r.created_at as "createdAt"
|
||||
from agent_task_sessions s join heartbeat_runs r on r.id = s.last_run_id
|
||||
and r.company_id = s.company_id and r.agent_id = s.agent_id
|
||||
where s.adapter_type = 'paperclip_runner' and r.native_issue_id is not null and r.native_session_id is not null
|
||||
${after ? sql`and s.id > ${after}::uuid` : sql``} order by s.id limit 100`);
|
||||
for (const selected of rows) {
|
||||
signal?.throwIfAborted();
|
||||
const source = await findNativeSessionResumeRun(db, { companyId: selected.companyId, agentId: selected.agentId,
|
||||
issueId: selected.issueId, normalizedSessionId: selected.sessionId,
|
||||
beforeCreatedAt: new Date(selected.createdAt), includeProgressBarrier: true });
|
||||
if (source) yield { selected, source };
|
||||
}
|
||||
if (rows.length < 100) return;
|
||||
after = rows.at(-1)!.id;
|
||||
}
|
||||
}
|
||||
|
||||
async function readPrivateSource(file: string, maxBytes: number): Promise<string | undefined> {
|
||||
const handle = await open(file, constants.O_RDONLY | constants.O_NOFOLLOW).catch((error: NodeJS.ErrnoException) => {
|
||||
if (error.code === "ENOENT") return undefined; throw error;
|
||||
});
|
||||
if (!handle) return undefined;
|
||||
try {
|
||||
const stat = await handle.stat();
|
||||
if (!stat.isFile() || stat.nlink !== 1 || stat.size > maxBytes) throw new Error("native_controller_migration_unsafe_source");
|
||||
return await handle.readFile("utf8");
|
||||
} finally { await handle.close(); }
|
||||
}
|
||||
|
||||
async function localPaths(source: NonNullable<Awaited<ReturnType<typeof findNativeSessionResumeRun>>>) {
|
||||
const execution = parseNativeExecutionInput(source.runnerProfileJson.nativeExecutionInput);
|
||||
const descriptor = nativeControllerMigrationDescriptor(execution, source.runnerProfileJson.nativeToolContractFingerprint);
|
||||
const root = await lstat(descriptor.root).catch((error: NodeJS.ErrnoException) => {
|
||||
if (error.code === "ENOENT") return undefined; throw error;
|
||||
});
|
||||
if (root && (!root.isDirectory() || root.isSymbolicLink())) throw new Error("native_controller_migration_unsafe_source");
|
||||
const control = join(descriptor.root, "control-plane");
|
||||
const directory = await lstat(control).catch((error: NodeJS.ErrnoException) => {
|
||||
if (error.code === "ENOENT") return undefined; throw error;
|
||||
});
|
||||
if (directory && (!directory.isDirectory() || directory.isSymbolicLink())) throw new Error("native_controller_migration_unsafe_source");
|
||||
return { ...descriptor, execution, controlFile: join(control, "control-plane-state.json") };
|
||||
}
|
||||
|
||||
export async function hasLocalNativeControllerObligations(db: Db): Promise<boolean> {
|
||||
const roots = await readdir(nativeHarnessStateBase()).catch((error: NodeJS.ErrnoException) => {
|
||||
if (error.code === "ENOENT") return [] as string[]; throw error;
|
||||
});
|
||||
if (!roots.length) return false;
|
||||
for await (const { selected, source } of retainedSources(db)) {
|
||||
const binding = { companyId: selected.companyId, issueId: selected.issueId, runId: source.id };
|
||||
if (readNativeControllerRecovery(source.runnerProfileJson.nativeControllerRecovery, source.runnerProfileJson.sessionCheckpoint)
|
||||
&& await hasNativeRunnerCheckpoint(db, binding)) continue;
|
||||
const paths = await localPaths(source);
|
||||
const present = await lstat(paths.controlFile).catch((error: NodeJS.ErrnoException) => {
|
||||
if (error.code === "ENOENT") return undefined; throw error;
|
||||
});
|
||||
if (present || await hasNativeRunnerCheckpoint(db, binding)) return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
export async function migrateNativeControllers(db: Db, input: { writersStopped: true; signal: AbortSignal }) {
|
||||
const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(offlineMigrationExecutionWhere()).limit(1);
|
||||
if (!input.writersStopped || active) throw new Error("native_controller_migration_requires_offline_writers");
|
||||
const counts = { inspected: 0, imported: 0, unchanged: 0, sourceAlreadyAbsent: 0 };
|
||||
for await (const { selected, source } of retainedSources(db, input.signal)) {
|
||||
counts.inspected++;
|
||||
const binding = { companyId: selected.companyId, issueId: selected.issueId, runId: source.id };
|
||||
const checkpoint = await loadNativeRunnerCheckpoint(db, binding);
|
||||
const metadata = readNativeControllerRecovery(source.runnerProfileJson.nativeControllerRecovery, source.runnerProfileJson.sessionCheckpoint);
|
||||
const snapshot = source.runnerProfileJson.sessionCheckpoint as { identity?: Record<string, unknown> } | null;
|
||||
if (checkpoint && metadata?.checkpoint?.runId === source.id
|
||||
&& metadata.checkpoint.runnerInstanceId === source.runnerInstanceId
|
||||
&& metadata.checkpoint.normalizedSessionId === selected.sessionId
|
||||
&& snapshot?.identity?.runId === source.id && snapshot.identity.companyId === selected.companyId
|
||||
&& snapshot.identity.agentId === selected.agentId && snapshot.identity.issueId === selected.issueId
|
||||
&& snapshot.identity.sessionId === selected.sessionId
|
||||
&& checkpoint.recoveryEvidence.runnerInstanceId === source.runnerInstanceId
|
||||
&& checkpoint.recoveryEvidence.normalizedSessionId === selected.sessionId) { counts.unchanged++; continue; }
|
||||
const paths = await localPaths(source);
|
||||
const legacyControllerState = checkpoint ? undefined : await readPrivateSource(paths.controlFile, 256 * 1024 * 1024);
|
||||
if (!checkpoint && legacyControllerState === undefined) { counts.sourceAlreadyAbsent++; continue; }
|
||||
let legacyRunnerArtifact: { version: string; digest: string } | undefined;
|
||||
if (legacyControllerState !== undefined && await readPrivateSource(join(paths.root, "provider-pack-authority.json"), 512 * 1024)) {
|
||||
const identity = JSON.parse(legacyControllerState).identity;
|
||||
const execution = paths.execution;
|
||||
const [admission] = await db.select({ workspace: sql<unknown>`${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'` })
|
||||
.from(heartbeatRuns).where(and(eq(heartbeatRuns.id, source.id), eq(heartbeatRuns.companyId, selected.companyId),
|
||||
eq(heartbeatRuns.agentId, selected.agentId)));
|
||||
const workspace = readNativeWorkspaceSyncReference(admission?.workspace);
|
||||
const remoteCwd = workspace?.remoteCwd ?? execution.workspace.cwd;
|
||||
if (typeof identity?.environmentLeaseId !== "string" || !remoteCwd
|
||||
|| (workspace && workspace.leaseId !== identity.environmentLeaseId)) {
|
||||
throw new Error("native_controller_migration_artifact_binding");
|
||||
}
|
||||
const unavailable = (): never => { throw new Error("native_controller_migration_artifact_unavailable"); };
|
||||
legacyRunnerArtifact = resolveRemoteProviderPackAuthority({ stateDirectory: paths.root, reattach: true,
|
||||
binding: { companyId: selected.companyId, agentId: selected.agentId,
|
||||
workspaceId: providerPackWorkspaceId(execution.binding.executionWorkspaceId, source.id, identity.environmentLeaseId),
|
||||
sessionId: selected.sessionId, runnerInstanceId: source.runnerInstanceId!,
|
||||
environmentLeaseId: identity.environmentLeaseId, remoteCwd,
|
||||
providerDigest: nativeSha256(execution.provider) },
|
||||
current: unavailable, runnerArtifact: unavailable }).runnerArtifact;
|
||||
}
|
||||
const supervisor = paths.execution.session.lifecyclePolicy.mode === "warm"
|
||||
? await readPrivateSource(paths.supervisorPath, 4 * 1024 * 1024) : undefined;
|
||||
input.signal.throwIfAborted();
|
||||
const result = await importNativeControllerRecovery(db, { ...binding, agentId: selected.agentId,
|
||||
normalizedSessionId: selected.sessionId, expectedConfigDigest: paths.configDigest,
|
||||
supervisor: supervisor === undefined ? undefined : JSON.parse(supervisor),
|
||||
offline: { writersStopped: true, legacyControllerState, legacyRunnerArtifact } });
|
||||
if (result.kind !== "imported" && result.kind !== "unchanged") {
|
||||
throw new Error(`native_controller_migration_blocked:${source.id}:${"reason" in result ? result.reason : result.kind}`);
|
||||
}
|
||||
counts[result.kind]++;
|
||||
}
|
||||
return counts;
|
||||
}
|
||||
@@ -1,6 +1,7 @@
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { sql } from "drizzle-orm";
|
||||
import { inspectNativeRunnerUpgradeHeader } from "./native-runner-checkpoint.js";
|
||||
import { inspectNativeRunnerUpgradeHeader, NATIVE_CHECKPOINT_HEADER_INSPECTION_MAX_BYTES,
|
||||
NATIVE_CHECKPOINT_INSPECTION_TOTAL_MAX_BYTES } from "./native-runner-checkpoint.js";
|
||||
|
||||
/** An explicit, paginated diagnostic. Never called by an idle/background scan.
|
||||
* This inventories selected retained task-session pointers, not all history.
|
||||
@@ -23,7 +24,8 @@ export async function readNativeControllerRecoveryInventory(db: Db, input: { aft
|
||||
r.native_session_id as "sessionId", r.runner_instance_id as "runnerId",
|
||||
case when pg_column_size(r.runner_profile_json->'nativeControllerRecovery') <= 4096
|
||||
then r.runner_profile_json->'nativeControllerRecovery' else null end as metadata,
|
||||
case when pg_column_size(c.material) <= 262144 and sum(pg_column_size(c.material)) over () <= 8388608
|
||||
case when pg_column_size(c.material) <= ${NATIVE_CHECKPOINT_HEADER_INSPECTION_MAX_BYTES}
|
||||
and sum(pg_column_size(c.material)) over () <= ${NATIVE_CHECKPOINT_INSPECTION_TOTAL_MAX_BYTES}
|
||||
then c.material else null end as material,
|
||||
(r.finished_at is null or r.status not in ('succeeded', 'failed', 'cancelled', 'interrupted', 'timed_out')
|
||||
or f.run_id is null or f.lease_owner is not null or f.recovery_state is not null
|
||||
|
||||
@@ -1,12 +1,13 @@
|
||||
import { and, eq, ne, sql } from "drizzle-orm";
|
||||
import { environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, nativeRunnerCheckpoints, type Db } from "@paperclipai/db";
|
||||
import type { AdapterExecutionTarget } from "@paperclipai/adapter-utils/execution-target";
|
||||
import type { DurablePrpCheckpoint, PersistedNativeSession } from "../../vendor/paperclip-runner/index.js";
|
||||
import { captureLegacyDurablePrpCheckpoint, parseNativeExecutionInput, type DurablePrpCheckpoint, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js";
|
||||
import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js";
|
||||
import { readRunCancellation } from "../run-cancellation.js";
|
||||
import { nativeSha256 } from "./canonical.js";
|
||||
import { findNativeSessionResumeRun, isNativeSessionId, isUnusedNativeSessionBootstrap, nativeSessionProviderEvidence } from "./native-session-resume.js";
|
||||
import { loadNativeRunnerCheckpoint } from "./native-runner-checkpoint.js";
|
||||
import { insertMigratedNativeRunnerCheckpoint, loadNativeRunnerCheckpoint } from "./native-runner-checkpoint.js";
|
||||
import { offlineMigrationExecutionWhere } from "../offline-migration-execution.js";
|
||||
import { redactRunnerApiPayloads } from "./runner-api-log-redaction.js";
|
||||
|
||||
export const NATIVE_CONTROLLER_RECOVERY_MAX_BYTES = 4 * 1024;
|
||||
@@ -223,10 +224,11 @@ export async function resolveNativeControllerRecovery(db: Db, input: {
|
||||
export async function importNativeControllerRecovery(db: Db, input: {
|
||||
companyId: string; agentId: string; issueId: string; runId: string; normalizedSessionId: string;
|
||||
expectedConfigDigest: string;
|
||||
supervisor: unknown;
|
||||
supervisor?: unknown;
|
||||
offline?: { writersStopped: true; legacyControllerState?: string; legacyRunnerArtifact?: { version: string; digest: string } };
|
||||
}): Promise<{ kind: "imported" | "unchanged" } | Exclude<NativeControllerRecoveryOutcome, { kind: "recoverable" } | { kind: "no_prior_execution" }>> {
|
||||
const supervisor = record(input.supervisor);
|
||||
if (supervisor.schema !== "paperclip.native-session-supervisor.v1" || supervisor.configDigest !== input.expectedConfigDigest) {
|
||||
if (input.supervisor !== undefined && (supervisor.schema !== "paperclip.native-session-supervisor.v1" || supervisor.configDigest !== input.expectedConfigDigest)) {
|
||||
return { kind: "unsupported", reason: "supervisor_compatibility" };
|
||||
}
|
||||
return db.transaction(async tx => {
|
||||
@@ -236,25 +238,48 @@ export async function importNativeControllerRecovery(db: Db, input: {
|
||||
eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.issueId, input.issueId),
|
||||
eq(nativeRunFinalizations.runId, input.runId))).for("update");
|
||||
if (!task || !owner) return { kind: "missing_evidence" as const, reason: "owner_binding" };
|
||||
if (owner.leaseOwner || task.executionRunId) {
|
||||
if (input.offline) {
|
||||
const [active] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(offlineMigrationExecutionWhere()).limit(1);
|
||||
if (!input.offline.writersStopped || active) return { kind: "missing_evidence" as const, reason: "offline_writers_required" };
|
||||
}
|
||||
const [clock] = await tx.execute<{ now: Date }>(sql`select clock_timestamp() as now`);
|
||||
const liveOwner = owner.leaseOwner && (!owner.leaseExpiresAt || owner.leaseExpiresAt > new Date(clock.now));
|
||||
if (input.offline ? liveOwner : owner.leaseOwner || task.executionRunId) {
|
||||
return owner.leaseExpiresAt && owner.leaseExpiresAt > new Date()
|
||||
? { kind: "waiting_for_owner" as const, deadline: owner.leaseExpiresAt }
|
||||
: { kind: "missing_evidence" as const, reason: "normal_ownership_recovery_required" };
|
||||
}
|
||||
if (owner.phase !== "committed" || !owner.resultId || owner.recoveryState) {
|
||||
if (!input.offline && (!(owner.phase === "committed" && owner.resultId) || owner.recoveryState)) {
|
||||
return { kind: "missing_evidence" as const, reason: "pending_finalization" };
|
||||
}
|
||||
const [run] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, input.runId),
|
||||
eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.agentId, input.agentId),
|
||||
eq(heartbeatRuns.nativeIssueId, input.issueId), eq(heartbeatRuns.nativeSessionId, input.normalizedSessionId))).for("update");
|
||||
const profile = record(run?.runnerProfileJson), snapshot = record(profile.sessionCheckpoint), identity = record(snapshot.identity);
|
||||
if (run?.runtimeMode !== "native" || run.status !== "succeeded" || !run.finishedAt
|
||||
if (run?.runtimeMode !== "native" || !(run.status === "succeeded" || input.offline && ["failed", "cancelled", "interrupted", "timed_out"].includes(run.status)) || !run.finishedAt
|
||||
|| identity.runId !== input.runId || identity.companyId !== input.companyId || identity.issueId !== input.issueId
|
||||
|| identity.agentId !== input.agentId || identity.sessionId !== input.normalizedSessionId
|
||||
|| nativeSha256(snapshot) !== nativeSha256(redactRunnerApiPayloads(supervisor.snapshot))) {
|
||||
|| input.supervisor !== undefined && nativeSha256(snapshot) !== nativeSha256(redactRunnerApiPayloads(supervisor.snapshot))) {
|
||||
return { kind: "missing_evidence" as const, reason: "stale_supervisor_snapshot" };
|
||||
}
|
||||
const checkpoint = await loadNativeRunnerCheckpoint(tx, input);
|
||||
if (input.supervisor === undefined) {
|
||||
try {
|
||||
const admitted = parseNativeExecutionInput(profile.nativeExecutionInput);
|
||||
if (!input.offline || admitted.session.lifecyclePolicy.mode !== "per_turn"
|
||||
|| admitted.binding.runId !== input.runId || admitted.binding.companyId !== input.companyId
|
||||
|| admitted.binding.agentId !== input.agentId || admitted.binding.issueId !== input.issueId) {
|
||||
return { kind: "missing_evidence" as const, reason: "admitted_per_turn_binding" };
|
||||
}
|
||||
} catch { return { kind: "unsupported" as const, reason: "admitted_input_format" }; }
|
||||
}
|
||||
let checkpoint = await loadNativeRunnerCheckpoint(tx, input);
|
||||
const importing = !checkpoint && input.offline?.legacyControllerState !== undefined;
|
||||
if (importing) {
|
||||
if (!run.runnerInstanceId) return { kind: "missing_evidence" as const, reason: "checkpoint_binding" };
|
||||
checkpoint = captureLegacyDurablePrpCheckpoint(input.offline!.legacyControllerState!, {
|
||||
runId: input.runId, runnerInstanceId: run.runnerInstanceId, normalizedSessionId: input.normalizedSessionId,
|
||||
}, input.offline!.legacyRunnerArtifact);
|
||||
}
|
||||
if (!checkpoint || checkpoint.recoveryEvidence.normalizedSessionId !== input.normalizedSessionId
|
||||
|| checkpoint.recoveryEvidence.runnerInstanceId !== run.runnerInstanceId) {
|
||||
return { kind: "missing_evidence" as const, reason: "checkpoint_binding" };
|
||||
@@ -265,10 +290,13 @@ export async function importNativeControllerRecovery(db: Db, input: {
|
||||
}, { runId: input.runId, runnerInstanceId: checkpoint.recoveryEvidence.runnerInstanceId, normalizedSessionId: input.normalizedSessionId });
|
||||
if (!readNativeControllerRecovery(metadata, snapshot)) return { kind: "unsupported" as const, reason: "metadata_format" };
|
||||
if (profile.nativeControllerRecovery) {
|
||||
return nativeSha256(profile.nativeControllerRecovery) === nativeSha256(metadata)
|
||||
? { kind: "unchanged" as const }
|
||||
: { kind: "missing_evidence" as const, reason: "existing_metadata_conflict" };
|
||||
if (nativeSha256(profile.nativeControllerRecovery) !== nativeSha256(metadata)) {
|
||||
return { kind: "missing_evidence" as const, reason: "existing_metadata_conflict" };
|
||||
}
|
||||
if (importing) await insertMigratedNativeRunnerCheckpoint(tx, input, checkpoint);
|
||||
return { kind: importing ? "imported" as const : "unchanged" as const };
|
||||
}
|
||||
if (importing) await insertMigratedNativeRunnerCheckpoint(tx, input, checkpoint);
|
||||
await tx.update(heartbeatRuns).set({ runnerProfileJson: { ...profile, nativeControllerRecovery: metadata }, updatedAt: new Date() })
|
||||
.where(eq(heartbeatRuns.id, input.runId));
|
||||
return { kind: "imported" as const };
|
||||
|
||||
@@ -0,0 +1,46 @@
|
||||
import { mkdtemp, readFile, rm, stat, symlink, writeFile } from "node:fs/promises";
|
||||
import { createWriteStream } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
import { tmpdir } from "node:os";
|
||||
import { pipeline } from "node:stream/promises";
|
||||
import tar from "tar-stream";
|
||||
import { afterEach, expect, it } from "vitest";
|
||||
import { captureNativeHarnessArchive, restoreNativeHarnessArchive } from "./native-harness-backup-archive.js";
|
||||
|
||||
const roots: string[] = [];
|
||||
afterEach(async () => { for (const root of roots.splice(0)) await rm(root, { recursive: true, force: true }); });
|
||||
async function fixture() { const root = await mkdtemp(join(tmpdir(), "native-backup-archive-test-")); roots.push(root); return root; }
|
||||
async function maliciousArchive(root: string, headers: Array<tar.Headers>) {
|
||||
const archive = tar.pack(), file = join(root, "malicious.tar");
|
||||
const written = pipeline(archive, createWriteStream(file));
|
||||
for (const header of headers) archive.entry(header, Buffer.alloc(0));
|
||||
archive.finalize(); await written; return file;
|
||||
}
|
||||
it.each([
|
||||
{ name: "../escaped", type: "file" as const },
|
||||
{ name: "/absolute", type: "file" as const },
|
||||
{ name: "link", type: "symlink" as const, linkname: "../outside" },
|
||||
{ name: "device", type: "character-device" as const },
|
||||
{ name: "parent/child", type: "file" as const },
|
||||
])("rejects unsafe archive entry $name", async header => {
|
||||
const root = await fixture(), file = await maliciousArchive(root, [header]);
|
||||
await expect(restoreNativeHarnessArchive(file, join(root, "destination"))).rejects.toThrow("native_harness_archive_invalid");
|
||||
await expect(stat(join(root, "escaped"))).rejects.toMatchObject({ code: "ENOENT" });
|
||||
});
|
||||
it("rejects duplicate paths instead of overwriting a restored file", async () => {
|
||||
const root = await fixture(), file = await maliciousArchive(root, [{ name: "manifest.json", type: "file" }, { name: "manifest.json", type: "file" }]);
|
||||
await expect(restoreNativeHarnessArchive(file, join(root, "destination"))).rejects.toThrow("native_harness_archive_invalid");
|
||||
});
|
||||
it("refuses a source symlink without reading its target", async () => {
|
||||
const root = await fixture(), outside = join(root, "outside");
|
||||
await writeFile(outside, "private external file");
|
||||
const { mkdir } = await import("node:fs/promises"); const source = join(root, "source"); await mkdir(source);
|
||||
await symlink(outside, join(source, "linked"));
|
||||
await expect(captureNativeHarnessArchive(source, join(root, "archive"))).rejects.toThrow("native_harness_archive_invalid");
|
||||
expect(await readFile(outside, "utf8")).toBe("private external file");
|
||||
});
|
||||
it("honors cancellation before walking the checkpoint tree", async () => {
|
||||
const root = await fixture();
|
||||
await expect(captureNativeHarnessArchive(root, join(root, "archive"), AbortSignal.abort(new Error("test cancelled"))))
|
||||
.rejects.toThrow();
|
||||
});
|
||||
@@ -0,0 +1,79 @@
|
||||
import { constants, createReadStream, createWriteStream } from "node:fs";
|
||||
import { chmod, lstat, mkdir, open, readdir } from "node:fs/promises";
|
||||
import { join, posix } from "node:path";
|
||||
import { pipeline } from "node:stream/promises";
|
||||
import tar from "tar-stream";
|
||||
|
||||
const MAX_BYTES = 256 * 1024 * 1024;
|
||||
const MAX_ENTRIES = 40_000;
|
||||
const fail = () => new Error("native_harness_archive_invalid");
|
||||
const safeName = (name: string) => name.length > 0 && !name.includes("\\")
|
||||
&& !/[\u0000-\u001f\u007f]/u.test(name) && !posix.isAbsolute(name)
|
||||
&& name.split("/").every(part => part !== ".." && part !== "." && part !== "");
|
||||
|
||||
export async function captureNativeHarnessArchive(root: string, target: string, signal?: AbortSignal) {
|
||||
if (!(await lstat(root)).isDirectory()) throw fail();
|
||||
const archive = tar.pack();
|
||||
const output = pipeline(archive, createWriteStream(target, { flags: "wx", mode: 0o600 }), { signal });
|
||||
void output.catch(() => undefined);
|
||||
let count = 0, bytes = 0;
|
||||
const visit = async (relative: string): Promise<void> => {
|
||||
for (const name of (await readdir(join(root, relative))).sort()) {
|
||||
signal?.throwIfAborted();
|
||||
const entry = relative ? `${relative}/${name}` : name;
|
||||
if (!safeName(entry) || ++count > MAX_ENTRIES) throw fail();
|
||||
const file = join(root, entry), stat = await lstat(file);
|
||||
if (stat.isDirectory()) {
|
||||
await new Promise<void>((resolve, reject) => archive.entry({ name: entry, type: "directory", mode: stat.mode & 0o777 },
|
||||
error => error ? reject(error) : resolve()));
|
||||
await visit(entry);
|
||||
} else if (stat.isFile()) {
|
||||
bytes += stat.size;
|
||||
if (bytes > MAX_BYTES) throw fail();
|
||||
const handle = await open(file, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK);
|
||||
try {
|
||||
const before = await handle.stat();
|
||||
if (!before.isFile() || before.ino !== stat.ino || before.dev !== stat.dev || before.size !== stat.size) throw fail();
|
||||
const stream = archive.entry({ name: entry, size: stat.size, type: "file", mode: stat.mode & 0o777 });
|
||||
await pipeline(handle.createReadStream({ autoClose: false }), stream, { signal });
|
||||
const after = await handle.stat();
|
||||
if (before.size !== after.size || before.mtimeMs !== after.mtimeMs || before.ctimeMs !== after.ctimeMs) throw fail();
|
||||
} finally { await handle.close(); }
|
||||
} else throw fail();
|
||||
}
|
||||
};
|
||||
try { await visit(""); archive.finalize(); await output; }
|
||||
catch (error) { archive.destroy(error as Error); await output.catch(() => undefined); throw error; }
|
||||
}
|
||||
|
||||
export async function restoreNativeHarnessArchive(source: string, target: string, signal?: AbortSignal) {
|
||||
await mkdir(target, { mode: 0o700 });
|
||||
const archive = tar.extract(), names = new Set<string>();
|
||||
const directories: Array<{ path: string; mode: number }> = [], directoryNames = new Set<string>();
|
||||
let bytes = 0;
|
||||
archive.on("entry", (header, body, next) => {
|
||||
body.on("error", () => undefined);
|
||||
void (async () => {
|
||||
signal?.throwIfAborted();
|
||||
const name = header.type === "directory" ? header.name.replace(/\/$/u, "") : header.name;
|
||||
if (!safeName(name) || names.has(name) || names.size >= MAX_ENTRIES
|
||||
|| !["directory", "file"].includes(header.type ?? "") || !Number.isSafeInteger(header.size ?? 0)
|
||||
|| (header.size ?? 0) < 0 || (bytes += header.size ?? 0) > MAX_BYTES) throw fail();
|
||||
names.add(name);
|
||||
const file = join(target, name), mode = (header.mode ?? 0o600) & 0o777;
|
||||
const parent = posix.dirname(name);
|
||||
if (parent !== "." && !directoryNames.has(parent)) throw fail();
|
||||
if (header.type === "directory") {
|
||||
if (header.size) throw fail();
|
||||
await mkdir(file, { mode: 0o700 }); directories.push({ path: file, mode }); directoryNames.add(name);
|
||||
for await (const _ of body) throw fail();
|
||||
} else {
|
||||
await pipeline(body, createWriteStream(file, { flags: "wx", mode }), { signal });
|
||||
await chmod(file, mode);
|
||||
}
|
||||
})().then(() => next(), error => archive.destroy(error as Error));
|
||||
});
|
||||
await pipeline(createReadStream(source), archive, { signal });
|
||||
if (!names.has("manifest.json")) throw fail();
|
||||
for (const directory of directories.reverse()) await chmod(directory.path, directory.mode);
|
||||
}
|
||||
@@ -2,7 +2,7 @@ import { notifyExternalConversationSettlement } from "../external-conversation-s
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { promises as fs } from "node:fs";
|
||||
import path from "node:path";
|
||||
import { and, asc, eq, gt, inArray, isNull, lt, or, sql } from "drizzle-orm";
|
||||
import { and, asc, eq, gt, gte, inArray, isNull, lt, or, sql } from "drizzle-orm";
|
||||
import {
|
||||
agentTaskSessions,
|
||||
environmentLeases,
|
||||
@@ -10,6 +10,7 @@ import {
|
||||
issues,
|
||||
nativeRunFinalizations,
|
||||
nativeRunnerCheckpoints,
|
||||
nativeHarnessBackups,
|
||||
type Db,
|
||||
} from "@paperclipai/db";
|
||||
import { logger } from "../../middleware/logger.js";
|
||||
@@ -18,6 +19,8 @@ import { nativeSessionScopeDigestIsBusy } from "./native-session-executor.js";
|
||||
import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
|
||||
import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
|
||||
import type { NativeHarnessBackupExpirationScope } from "../native-harness-backup-expiration-notifications.js";
|
||||
import { deleteRetiredNativeHarnessBackup, durableNativeHarnessStorage } from "./native-harness-backup-storage.js";
|
||||
import type { StorageService } from "../../storage/types.js";
|
||||
|
||||
export const NATIVE_HARNESS_BACKUP_EXPIRATION_INTERVAL_MS = 60 * 60 * 1000;
|
||||
export const NATIVE_HARNESS_BACKUP_EXPIRATION_STARTUP_DELAY_MS = 2 * 60 * 1000;
|
||||
@@ -51,6 +54,7 @@ export type NativeHarnessBackupExpirationOptions = {
|
||||
stateBase?: string;
|
||||
shouldContinue?: () => boolean;
|
||||
isSessionBusy?: (scopeDigest: string) => boolean;
|
||||
storage?: StorageService;
|
||||
};
|
||||
|
||||
export type NativeHarnessBackupExpirationResult = {
|
||||
@@ -291,6 +295,67 @@ export function nativeHarnessBackupExpirationService(
|
||||
let evidenceDueAt = 0;
|
||||
let passDueAt = Number.POSITIVE_INFINITY;
|
||||
let checkpointCursor: string | null = null;
|
||||
let durableBackupCursor: string | null = null;
|
||||
let checkpointsEnumerated = false;
|
||||
|
||||
async function expireDurableBackups(options: {
|
||||
closedBefore: Date; dryRun: boolean; maxEntries: number; maxRemovals: number; shouldContinue: () => boolean;
|
||||
}, result: NativeHarnessBackupExpirationResult) {
|
||||
const storage = defaults.storage ?? durableNativeHarnessStorage();
|
||||
if (!storage) return;
|
||||
const candidates = await db.select({ id: nativeHarnessBackups.id, scopeDigest: nativeHarnessBackups.scopeDigest,
|
||||
runId: nativeHarnessBackups.runId, survivingRunId: heartbeatRuns.id,
|
||||
sessionId: heartbeatRuns.nativeSessionId, status: nativeHarnessBackups.status })
|
||||
.from(nativeHarnessBackups).leftJoin(heartbeatRuns, eq(heartbeatRuns.id, nativeHarnessBackups.runId))
|
||||
.where(and(durableBackupCursor ? gt(nativeHarnessBackups.id, durableBackupCursor) : undefined,
|
||||
or(eq(nativeHarnessBackups.status, "retired"), lt(nativeHarnessBackups.createdAt, options.closedBefore))))
|
||||
.orderBy(asc(nativeHarnessBackups.id)).limit(options.maxEntries + 1);
|
||||
let processed = 0;
|
||||
for (const candidate of candidates) {
|
||||
if (!options.shouldContinue() || processed++ >= options.maxEntries || result.expired >= options.maxRemovals) {
|
||||
result.truncated = true; return;
|
||||
}
|
||||
durableBackupCursor = candidate.id;
|
||||
if (candidate.survivingRunId && !candidate.sessionId) continue;
|
||||
try {
|
||||
const retired = candidate.status === "retired" || await db.transaction(async tx => {
|
||||
if (!candidate.survivingRunId) {
|
||||
result.candidates++;
|
||||
if (options.dryRun) return false;
|
||||
await tx.update(nativeHarnessBackups).set({ status: "retired", updatedAt: new Date() })
|
||||
.where(eq(nativeHarnessBackups.id, candidate.id));
|
||||
return true;
|
||||
}
|
||||
const blocker = await sessionRetentionBlocker(tx, { scopeDigest: candidate.scopeDigest, checkpoint: true,
|
||||
closedBefore: options.closedBefore,
|
||||
identity: { runId: candidate.runId, normalizedSessionId: candidate.sessionId!, environmentLeaseId: null } });
|
||||
if ("reason" in blocker) return false;
|
||||
const [recent] = await tx.select({ id: nativeHarnessBackups.id }).from(nativeHarnessBackups).where(and(
|
||||
eq(nativeHarnessBackups.companyId, blocker.companyId), eq(nativeHarnessBackups.scopeDigest, candidate.scopeDigest),
|
||||
gte(nativeHarnessBackups.createdAt, options.closedBefore))).limit(1);
|
||||
if (recent) return false;
|
||||
result.candidates++;
|
||||
if (options.dryRun) return false;
|
||||
await notifyExternalConversationSettlement(tx, { companyId: blocker.companyId });
|
||||
await tx.update(nativeRunFinalizations).set({ controllerGeneration: sql`${nativeRunFinalizations.controllerGeneration} + 1` })
|
||||
.where(inArray(nativeRunFinalizations.runId, blocker.runIds));
|
||||
await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${`${blocker.companyId}:${candidate.scopeDigest}`}, 0))`);
|
||||
const cleared = await tx.delete(agentTaskSessions).where(and(eq(agentTaskSessions.companyId, blocker.companyId),
|
||||
eq(agentTaskSessions.agentId, blocker.agentId), or(inArray(agentTaskSessions.lastRunId, blocker.runIds),
|
||||
sql`${agentTaskSessions.sessionParamsJson}->>'sessionId' = ${candidate.sessionId}`))).returning({ id: agentTaskSessions.id });
|
||||
result.clearedTaskSessions += cleared.length;
|
||||
await tx.update(nativeHarnessBackups).set({ status: "retired", updatedAt: new Date() }).where(and(
|
||||
eq(nativeHarnessBackups.companyId, blocker.companyId), eq(nativeHarnessBackups.scopeDigest, candidate.scopeDigest)));
|
||||
return true;
|
||||
});
|
||||
if (retired && !options.dryRun && await deleteRetiredNativeHarnessBackup(db, storage, candidate.id, AbortSignal.timeout(60_000))) result.expired++;
|
||||
} catch (err) {
|
||||
result.failed++;
|
||||
logger.warn({ err, backupId: candidate.id }, "durable native harness backup expiration skipped an object");
|
||||
}
|
||||
}
|
||||
durableBackupCursor = null;
|
||||
}
|
||||
|
||||
async function expireCheckpoints(options: {
|
||||
closedBefore: Date; dryRun: boolean; maxEntries: number; maxRemovals: number; shouldContinue: () => boolean;
|
||||
@@ -704,13 +769,20 @@ export function nativeHarnessBackupExpirationService(
|
||||
const result = emptyResult();
|
||||
const shouldContinue = options.shouldContinue ?? defaults.shouldContinue ?? (() => true);
|
||||
if (!shouldContinue()) return { ...result, truncated: true };
|
||||
await expireCheckpoints({
|
||||
const pass = {
|
||||
closedBefore: new Date((options.now ?? defaults.now ?? new Date()).getTime() - closedForDays * DAY_MS),
|
||||
dryRun: options.dryRun ?? defaults.dryRun ?? false,
|
||||
maxEntries: Math.max(1, options.maxEntries ?? defaults.maxEntries ?? 200),
|
||||
maxRemovals: Math.max(1, options.maxRemovals ?? defaults.maxRemovals ?? 50),
|
||||
shouldContinue,
|
||||
}, result);
|
||||
};
|
||||
if (!checkpointsEnumerated) {
|
||||
await expireCheckpoints(pass, result);
|
||||
if (result.truncated) return result;
|
||||
checkpointsEnumerated = true;
|
||||
}
|
||||
await expireDurableBackups(pass, result);
|
||||
if (!result.truncated) checkpointsEnumerated = false;
|
||||
return result;
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,76 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import { lstat, readFile, readdir } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { environmentLeases, heartbeatRuns, nativeHarnessBackups, type Db } from "@paperclipai/db";
|
||||
import type { StorageService } from "../../storage/types.js";
|
||||
import { nativeHarnessStateBase, readNativeHarnessBackupStamp, verifyNativeHarnessBackupStamp } from "./native-harness-backup-stamp.js";
|
||||
import { persistNativeHarnessBackup, verifyDurableNativeHarnessBackup } from "./native-harness-backup-storage.js";
|
||||
import { offlineMigrationExecutionWhere } from "../offline-migration-execution.js";
|
||||
|
||||
export async function hasLocalNativeHarnessObligations(db: Db): Promise<boolean> {
|
||||
const rows = await db.execute(sql`select 1 from agent_task_sessions s
|
||||
join heartbeat_runs r on r.id = s.last_run_id and r.company_id = s.company_id and r.agent_id = s.agent_id
|
||||
join environment_leases e on e.company_id = r.company_id and e.issue_id = r.native_issue_id
|
||||
and e.metadata->'nativeHarnessBackup'->>'normalizedSessionId' = r.native_session_id::text
|
||||
where s.adapter_type = 'paperclip_runner'
|
||||
and e.metadata->'nativeHarnessBackup'->>'schema' = 'paperclip.native-harness-backup-stamp.v2'
|
||||
and not exists (select 1 from native_harness_backups b where b.company_id = r.company_id
|
||||
and b.scope_digest = e.metadata->'nativeHarnessBackup'->>'sessionScopeSha256' and b.status = 'current')
|
||||
limit 1`);
|
||||
return rows.length > 0;
|
||||
}
|
||||
|
||||
export async function migrateNativeHarnessBackups(db: Db, input: {
|
||||
storage: StorageService; writersStopped: true; signal: AbortSignal;
|
||||
}): Promise<number> {
|
||||
if (!input.writersStopped) throw new Error("native_harness_migration_requires_offline_writers");
|
||||
const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(offlineMigrationExecutionWhere()).limit(1);
|
||||
if (active) throw new Error("native_harness_migration_requires_offline_writers");
|
||||
const base = nativeHarnessStateBase();
|
||||
const directories = await readdir(base).catch((error: NodeJS.ErrnoException) => {
|
||||
if (error.code === "ENOENT") return [] as string[]; throw error;
|
||||
});
|
||||
let count = 0;
|
||||
for (const scopeDigest of directories.filter(name => /^[a-f0-9]{64}$/.test(name)).sort()) {
|
||||
input.signal.throwIfAborted();
|
||||
const root = join(base, scopeDigest, "failover-backups", "current");
|
||||
const stat = await lstat(root).catch((error: NodeJS.ErrnoException) => {
|
||||
if (error.code === "ENOENT") return null; throw error;
|
||||
});
|
||||
if (!stat) continue;
|
||||
if (!stat.isDirectory() || stat.isSymbolicLink()) throw new Error("native_harness_migration_unsafe_source");
|
||||
const manifestBytes = await readFile(join(root, "manifest.json"));
|
||||
const manifest = JSON.parse(manifestBytes.toString("utf8"));
|
||||
const state = JSON.parse(await readFile(join(root, "runner", "runner-state.json"), "utf8"));
|
||||
if (state.lifecycle !== "suspended" || typeof state.runId !== "string"
|
||||
|| manifest.schema !== "paperclip.native-harness-backup.v1") throw new Error("native_harness_migration_source_unsettled");
|
||||
const [run] = await db.select({ companyId: heartbeatRuns.companyId, issueId: heartbeatRuns.nativeIssueId,
|
||||
sessionId: heartbeatRuns.nativeSessionId, runnerId: heartbeatRuns.runnerInstanceId }).from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.id, state.runId));
|
||||
if (!run?.issueId || run.sessionId !== manifest.normalizedSessionId || run.runnerId !== manifest.runnerInstanceId) {
|
||||
throw new Error("native_harness_migration_binding_mismatch");
|
||||
}
|
||||
const [current] = await db.select().from(nativeHarnessBackups).where(and(eq(nativeHarnessBackups.companyId, run.companyId),
|
||||
eq(nativeHarnessBackups.scopeDigest, scopeDigest), eq(nativeHarnessBackups.status, "current")));
|
||||
if (current) {
|
||||
await verifyDurableNativeHarnessBackup({ db, storage: input.storage, companyId: run.companyId,
|
||||
scopeDigest, manifestSha256: current.manifestSha256, signal: input.signal });
|
||||
count++; continue;
|
||||
}
|
||||
const manifestSha256 = `sha256:${createHash("sha256").update(manifestBytes).digest("hex")}`;
|
||||
const leases = await db.select({ stamp: sql<unknown>`${environmentLeases.metadata}->'nativeHarnessBackup'`,
|
||||
providerLeaseId: environmentLeases.providerLeaseId }).from(environmentLeases).where(and(
|
||||
eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.issueId, run.issueId),
|
||||
sql`${environmentLeases.metadata}->'nativeHarnessBackup'->>'manifestSha256' = ${manifestSha256}`));
|
||||
const valid = leases.some(lease => {
|
||||
const stamp = readNativeHarnessBackupStamp(lease.stamp, lease.providerLeaseId ?? "");
|
||||
return stamp?.sessionScopeSha256 === scopeDigest && verifyNativeHarnessBackupStamp(stamp, lease.providerLeaseId!);
|
||||
});
|
||||
if (!valid) throw new Error("native_harness_migration_backup_unverified");
|
||||
await persistNativeHarnessBackup({ db, storage: input.storage, root, scopeDigest, signal: input.signal,
|
||||
owner: { kind: "offline_migration", writersStopped: true, companyId: run.companyId, issueId: run.issueId, runId: state.runId } });
|
||||
count++;
|
||||
}
|
||||
return count;
|
||||
}
|
||||
@@ -142,12 +142,12 @@ export function createNativeHarnessBackupStamp(input: {
|
||||
};
|
||||
}
|
||||
|
||||
export function verifyNativeHarnessBackupStamp(
|
||||
export function readNativeHarnessBackupStamp(
|
||||
value: unknown,
|
||||
expectedProviderLeaseId: string,
|
||||
): boolean {
|
||||
if (!expectedProviderLeaseId) return false;
|
||||
if (!value || typeof value !== "object" || Array.isArray(value)) return false;
|
||||
): NativeHarnessBackupStamp | null {
|
||||
if (!expectedProviderLeaseId) return null;
|
||||
if (!value || typeof value !== "object" || Array.isArray(value)) return null;
|
||||
const stamp = value as Record<string, unknown>;
|
||||
if (
|
||||
stamp.schema !== "paperclip.native-harness-backup-stamp.v2" ||
|
||||
@@ -160,9 +160,17 @@ export function verifyNativeHarnessBackupStamp(
|
||||
!stamp.sourceProviderLeaseId ||
|
||||
stamp.authorizedProviderLeaseId !== expectedProviderLeaseId ||
|
||||
typeof stamp.manifestSha256 !== "string" ||
|
||||
!stamp.manifestSha256.startsWith("sha256:")
|
||||
!/^sha256:[a-f0-9]{64}$/.test(stamp.manifestSha256) ||
|
||||
!/^[a-f0-9]{64}$/.test(stamp.sessionScopeSha256) ||
|
||||
typeof stamp.completedAt !== "string" || !Number.isFinite(Date.parse(stamp.completedAt))
|
||||
)
|
||||
return false;
|
||||
return null;
|
||||
return stamp as unknown as NativeHarnessBackupStamp;
|
||||
}
|
||||
|
||||
export function verifyNativeHarnessBackupStamp(value: unknown, expectedProviderLeaseId: string): boolean {
|
||||
const stamp = readNativeHarnessBackupStamp(value, expectedProviderLeaseId);
|
||||
if (!stamp) return false;
|
||||
const root = stateRootFromDigest(stamp.sessionScopeSha256);
|
||||
if (!root || !isRealDirectory(root)) return false;
|
||||
const backupRoot = resolve(root, "failover-backups");
|
||||
|
||||
@@ -0,0 +1,175 @@
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { chmod, mkdir, readFile, rename, stat, writeFile } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { agentTaskSessions, environmentLeases, heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations } from "@paperclipai/db";
|
||||
import { afterEach, expect, it, vi } from "vitest";
|
||||
import { describeEmbeddedPostgres, nativeRestartFixture } from "../../__tests__/helpers/native-restart.js";
|
||||
import { createLocalDiskStorageProvider } from "../../storage/local-disk-provider.js";
|
||||
import { createStorageService } from "../../storage/service.js";
|
||||
import { nativeHarnessBackupOwner, persistNativeHarnessBackup, restoreDurableNativeHarnessBackup,
|
||||
verifyDurableNativeHarnessBackup } from "./native-harness-backup-storage.js";
|
||||
import { removeNativeHarnessBackup } from "./native-harness-backup-cleanup.js";
|
||||
import { buildNativeHarnessBackupManifest } from "./native-session-executor.js";
|
||||
import { createNativeHarnessBackupStamp } from "./native-harness-backup-stamp.js";
|
||||
import { hasLocalNativeHarnessObligations, migrateNativeHarnessBackups } from "./native-harness-backup-migration.js";
|
||||
import { nativeHarnessBackupExpirationService } from "./native-harness-backup-expiration.js";
|
||||
|
||||
describeEmbeddedPostgres("durable native provider backups", () => {
|
||||
const environment = nativeRestartFixture();
|
||||
const ownedHomes: string[] = [];
|
||||
afterEach(async () => {
|
||||
for (const home of ownedHomes.splice(0)) await removeNativeHarnessBackup(home);
|
||||
vi.unstubAllEnvs();
|
||||
});
|
||||
async function fixture(provider = "codex") {
|
||||
vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "51".repeat(32));
|
||||
const f = await environment.seedRun(randomUUID());
|
||||
const { companyId, runtimeRoot, successor } = environment.current;
|
||||
await f.db.update(nativeRunFinalizations).set({ controllerBootId: successor.bootId, controllerGeneration: 1,
|
||||
leaseOwner: "backup-writer", leaseExpiresAt: new Date(Date.now() + 120_000) }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
const owner = await nativeHarnessBackupOwner(f.db, { companyId, issueId: f.issueId, runId: f.runId, bootId: successor.bootId });
|
||||
const home = join(runtimeRoot, randomUUID()), root = join(home, "failover-backups", "current");
|
||||
ownedHomes.push(home);
|
||||
await mkdir(join(root, provider, "sessions"), { recursive: true });
|
||||
await mkdir(join(root, "runner"));
|
||||
const identity = { normalizedSessionId: randomUUID(), providerSessionId: randomUUID(), provider };
|
||||
await writeFile(join(root, provider, "sessions", "history.json"), JSON.stringify(identity), { mode: 0o600 });
|
||||
await writeFile(join(root, "runner", "runner-state.json"), JSON.stringify({ runId: f.runId, lifecycle: "suspended" }), { mode: 0o600 });
|
||||
const manifest = JSON.stringify({ schema: "paperclip.native-harness-backup.v1", ...identity, directories: [provider, "runner"] });
|
||||
await writeFile(join(root, "manifest.json"), manifest, { mode: 0o600 });
|
||||
await chmod(join(root, provider, "sessions"), 0o500);
|
||||
const sessionScopeId = randomUUID(), scopeDigest = createHash("sha256").update(sessionScopeId).digest("hex");
|
||||
const storage = createStorageService(createLocalDiskStorageProvider(join(runtimeRoot, "durable-objects")));
|
||||
return { ...f, home, root, companyId, owner, scopeDigest, sessionScopeId, storage, identity,
|
||||
manifestSha256: `sha256:${createHash("sha256").update(manifest).digest("hex")}` };
|
||||
}
|
||||
it.each(["codex", "claude", "opencode", "grok", "new-provider"])("restores %s session bytes after deleting the entire controller home", async provider => {
|
||||
const f = await fixture(provider);
|
||||
await persistNativeHarnessBackup(f);
|
||||
const [row] = await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest));
|
||||
expect(row.status).toBe("current");
|
||||
const object = await f.storage.getObject(f.companyId, row.reference.objectKey as string);
|
||||
const parts: Buffer[] = []; for await (const part of object.stream) parts.push(part);
|
||||
expect(Buffer.concat(parts).includes(Buffer.from(f.identity.providerSessionId))).toBe(false);
|
||||
await removeNativeHarnessBackup(f.home);
|
||||
const replacement = join(environment.current.runtimeRoot, randomUUID(), "failover-backups");
|
||||
ownedHomes.push(replacement);
|
||||
expect(await restoreDurableNativeHarnessBackup({ ...f, backupRoot: replacement })).toBe(true);
|
||||
expect(JSON.parse(await readFile(join(replacement, "current", provider, "sessions", "history.json"), "utf8"))).toEqual(f.identity);
|
||||
expect((await stat(join(replacement, "current", provider, "sessions"))).mode & 0o777).toBe(0o500);
|
||||
expect(await verifyDurableNativeHarnessBackup(f)).toBe(true);
|
||||
});
|
||||
it("keeps the last published backup when object storage fails", async () => {
|
||||
const f = await fixture(); await persistNativeHarnessBackup(f);
|
||||
const [before] = await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest));
|
||||
const put = vi.spyOn(f.storage, "putFile").mockRejectedValue(new Error("storage unavailable"));
|
||||
await expect(persistNativeHarnessBackup(f)).rejects.toThrow("storage unavailable"); put.mockRestore();
|
||||
const rows = await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest));
|
||||
expect(rows.filter(row => row.status === "current").map(row => row.id)).toEqual([before.id]);
|
||||
expect(rows.filter(row => row.status === "pending")).toHaveLength(1);
|
||||
expect(await readFile(join(f.root, "manifest.json"), "utf8")).toContain("native-harness-backup");
|
||||
});
|
||||
it("fences an old controller whose upload finishes after ownership changes", async () => {
|
||||
const f = await fixture(); await persistNativeHarnessBackup(f);
|
||||
const [before] = await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest));
|
||||
const put = f.storage.putFile.bind(f.storage);
|
||||
f.storage.putFile = async input => {
|
||||
const result = await put(input);
|
||||
await f.db.update(nativeRunFinalizations).set({ controllerBootId: randomUUID(), controllerGeneration: 2 })
|
||||
.where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
return result;
|
||||
};
|
||||
await expect(persistNativeHarnessBackup(f)).rejects.toThrow("native_runner_checkpoint_owner_lost");
|
||||
const rows = await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest));
|
||||
expect(rows.filter(row => row.status === "current").map(row => row.id)).toEqual([before.id]);
|
||||
});
|
||||
it("rejects corrupt objects and cross-company lookups without publishing a local cache", async () => {
|
||||
const f = await fixture(); await persistNativeHarnessBackup(f);
|
||||
const replacement = join(environment.current.runtimeRoot, randomUUID(), "failover-backups");
|
||||
expect(await restoreDurableNativeHarnessBackup({ ...f, companyId: randomUUID(), backupRoot: replacement })).toBe(false);
|
||||
const [row] = await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest));
|
||||
await f.storage.deleteObject(f.companyId, row.reference.objectKey as string);
|
||||
await f.storage.putFile({ companyId: f.companyId, namespace: "native-harness-backups", objectKey: row.reference.objectKey as string,
|
||||
originalFilename: null, contentType: "application/octet-stream", body: Buffer.from("corrupted") });
|
||||
await expect(restoreDurableNativeHarnessBackup({ ...f, backupRoot: replacement })).rejects.toThrow("hash_mismatch");
|
||||
await expect(stat(join(replacement, "current"))).rejects.toMatchObject({ code: "ENOENT" });
|
||||
});
|
||||
async function settle(f: { db: Awaited<ReturnType<typeof fixture>>["db"]; runId: string; issueId: string }, closedAt = new Date()) {
|
||||
await f.db.update(heartbeatRuns).set({ status: "failed", finishedAt: closedAt }).where(eq(heartbeatRuns.id, f.runId));
|
||||
await f.db.update(nativeRunFinalizations).set({ phase: "terminal_failure", leaseOwner: null, leaseExpiresAt: null })
|
||||
.where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
await f.db.update(issues).set({ status: "done", executionRunId: null, completedAt: closedAt }).where(eq(issues.id, f.issueId));
|
||||
}
|
||||
it("migrates a lease-verified local backup once and keeps its selected session intact", async () => {
|
||||
const f = await fixture("codex-home");
|
||||
const base = join(environment.current.runtimeRoot, randomUUID()); ownedHomes.push(base); await mkdir(base);
|
||||
await rename(f.home, join(base, f.scopeDigest));
|
||||
const root = join(base, f.scopeDigest, "failover-backups", "current");
|
||||
vi.stubEnv("PAPERCLIP_RUNNER_STATE_DIR", base);
|
||||
const runnerInstanceId = randomUUID(), providerLeaseId = randomUUID();
|
||||
const execution = { provider: { kind: "codex" }, binding: { companyId: f.companyId, issueId: f.issueId,
|
||||
runId: f.runId, agentId: environment.current.agentId, executionWorkspaceId: randomUUID() },
|
||||
workspace: { cwd: "/workspace", repoUrl: null, repoRef: null, branchName: null },
|
||||
session: { normalizedSessionId: f.identity.normalizedSessionId, driverKind: "codex_app_server",
|
||||
protocolVersion: 1, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } } } as never;
|
||||
const manifest = buildNativeHarnessBackupManifest({ backupRoot: root, execution, runnerInstanceId,
|
||||
providerSessionIdentity: { providerSessionId: f.identity.providerSessionId }, sourceProviderLeaseId: providerLeaseId });
|
||||
await writeFile(join(root, "manifest.json"), JSON.stringify(manifest));
|
||||
const stamp = createNativeHarnessBackupStamp({ manifestPath: join(root, "manifest.json"), sessionScopeId: f.sessionScopeId,
|
||||
authorizedProviderLeaseId: providerLeaseId, runnerInstanceId, normalizedSessionId: f.identity.normalizedSessionId, completedAt: manifest.completedAt });
|
||||
await f.db.update(heartbeatRuns).set({ nativeSessionId: f.identity.normalizedSessionId, runnerInstanceId }).where(eq(heartbeatRuns.id, f.runId));
|
||||
await f.db.insert(environmentLeases).values({ companyId: f.companyId, issueId: f.issueId, heartbeatRunId: f.runId,
|
||||
providerLeaseId, status: "expired", metadata: { nativeHarnessBackup: stamp } });
|
||||
const [selected] = await f.db.insert(agentTaskSessions).values({ companyId: f.companyId, agentId: environment.current.agentId,
|
||||
adapterType: "paperclip_runner", taskKey: f.issueId, lastRunId: f.runId,
|
||||
sessionParamsJson: { sessionId: f.identity.normalizedSessionId } }).returning();
|
||||
expect(await hasLocalNativeHarnessObligations(f.db)).toBe(true);
|
||||
await expect(migrateNativeHarnessBackups(f.db, { storage: f.storage, writersStopped: true, signal: AbortSignal.timeout(10_000) }))
|
||||
.rejects.toThrow("native_harness_migration_requires_offline_writers");
|
||||
const ownedRuns = await f.db.select({ id: heartbeatRuns.id, issueId: heartbeatRuns.nativeIssueId }).from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.companyId, f.companyId));
|
||||
for (const run of ownedRuns) await settle({ ...f, runId: run.id, issueId: run.issueId! });
|
||||
await f.db.update(nativeRunFinalizations).set({ recoveryState: "blocked" }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
const [preservedOwner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
const [queued] = await f.db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: environment.current.agentId,
|
||||
status: "queued", invocationSource: "on_demand", issueId: f.issueId }).returning();
|
||||
await f.db.update(issues).set({ executionRunId: queued.id }).where(eq(issues.id, f.issueId));
|
||||
expect(await migrateNativeHarnessBackups(f.db, { storage: f.storage, writersStopped: true, signal: AbortSignal.timeout(10_000) })).toBe(1);
|
||||
expect(await migrateNativeHarnessBackups(f.db, { storage: f.storage, writersStopped: true, signal: AbortSignal.timeout(10_000) })).toBe(1);
|
||||
expect(await hasLocalNativeHarnessObligations(f.db)).toBe(false);
|
||||
expect(await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest))).toHaveLength(1);
|
||||
expect((await f.db.select().from(agentTaskSessions).where(eq(agentTaskSessions.id, selected.id)))[0]).toEqual(selected);
|
||||
expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toEqual(preservedOwner);
|
||||
expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, queued.id)))[0]).toEqual(queued);
|
||||
expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0].executionRunId).toBe(queued.id);
|
||||
expect(JSON.parse(await readFile(join(root, "manifest.json"), "utf8"))).toEqual(manifest);
|
||||
});
|
||||
it("retains live work and retries an uncertain object deletion through the existing startup retention pass", async () => {
|
||||
const f = await fixture(); await persistNativeHarnessBackup(f);
|
||||
const old = new Date(Date.now() - 10 * 86_400_000);
|
||||
await f.db.update(nativeHarnessBackups).set({ createdAt: old }).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest));
|
||||
const retention = nativeHarnessBackupExpirationService(f.db, { storage: f.storage });
|
||||
expect((await retention.tickCheckpoints()).expired).toBe(0);
|
||||
await settle(f, old);
|
||||
const drop = vi.spyOn(f.storage, "deleteObject").mockRejectedValueOnce(new Error("delete reply lost"));
|
||||
expect((await retention.tickCheckpoints()).failed).toBe(1);
|
||||
expect((await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest)))[0].status).toBe("retired");
|
||||
expect((await retention.tickCheckpoints()).expired).toBe(1);
|
||||
expect(await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest))).toHaveLength(0);
|
||||
expect(drop).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
it("keeps deleted-source objects unchanged during a retention dry run", async () => {
|
||||
const f = await fixture(); await persistNativeHarnessBackup(f);
|
||||
const old = new Date(Date.now() - 10 * 86_400_000);
|
||||
await f.db.update(nativeHarnessBackups).set({ createdAt: old, runId: randomUUID() })
|
||||
.where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest));
|
||||
const [before] = await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest));
|
||||
const retention = nativeHarnessBackupExpirationService(f.db, { storage: f.storage });
|
||||
expect((await retention.tickCheckpoints({ dryRun: true })).expired).toBe(0);
|
||||
expect((await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.id, before.id)))[0]).toEqual(before);
|
||||
expect(await verifyDurableNativeHarnessBackup(f)).toBe(true);
|
||||
expect((await retention.tickCheckpoints()).expired).toBe(1);
|
||||
expect(await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.id, before.id))).toHaveLength(0);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,221 @@
|
||||
import { createCipheriv, createDecipheriv, createHash, randomBytes, randomUUID } from "node:crypto";
|
||||
import { createReadStream, createWriteStream } from "node:fs";
|
||||
import { mkdtemp, readFile, rename } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import os from "node:os";
|
||||
import { pipeline } from "node:stream/promises";
|
||||
import { Transform } from "node:stream";
|
||||
import { and, eq, inArray, sql } from "drizzle-orm";
|
||||
import { heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, type Db } from "@paperclipai/db";
|
||||
import { createLocalEncryptedCipher } from "../../secrets/local-encrypted-provider.js";
|
||||
import type { StorageService } from "../../storage/types.js";
|
||||
import { assertNativeRunnerCheckpointOwner, type NativeRunnerCheckpointOwner } from "./native-runner-checkpoint.js";
|
||||
import { loadConfig } from "../../config.js";
|
||||
import { getStorageService } from "../../storage/index.js";
|
||||
import { readNativeHarnessBackupStamp, verifyNativeHarnessBackupStamp } from "./native-harness-backup-stamp.js";
|
||||
import { captureNativeHarnessArchive, restoreNativeHarnessArchive } from "./native-harness-backup-archive.js";
|
||||
import { removeNativeHarnessBackup } from "./native-harness-backup-cleanup.js";
|
||||
import { offlineMigrationExecutionWhere } from "../offline-migration-execution.js";
|
||||
|
||||
const SCHEMA = "paperclip.native-harness-object.v1";
|
||||
type Binding = { companyId: string; scopeDigest: string; id: string };
|
||||
type Reference = Binding & {
|
||||
schema: typeof SCHEMA; objectKey: string; sha256: string; byteSize: number;
|
||||
material: Record<string, unknown>; manifestSha256: string;
|
||||
};
|
||||
type Row = typeof nativeHarnessBackups.$inferSelect;
|
||||
type OfflineOwner = { kind: "offline_migration"; companyId: string; issueId: string; runId: string; writersStopped: true };
|
||||
const fail = (reason: string): never => { throw new Error(`native_harness_storage_${reason}`); };
|
||||
const aad = (binding: Binding) => Buffer.from(JSON.stringify([SCHEMA, binding.companyId, binding.scopeDigest, binding.id]));
|
||||
const scope = (binding: Pick<Binding, "companyId" | "scopeDigest">) => and(
|
||||
eq(nativeHarnessBackups.companyId, binding.companyId), eq(nativeHarnessBackups.scopeDigest, binding.scopeDigest));
|
||||
|
||||
export function durableNativeHarnessStorage(): StorageService | null {
|
||||
if (loadConfig().storageProvider === "s3") return getStorageService();
|
||||
if (process.env.PAPERCLIP_EPHEMERAL_SERVER === "true") fail("durable_storage_required");
|
||||
return null;
|
||||
}
|
||||
|
||||
export async function verifyRecoverableNativeHarnessBackup(input: {
|
||||
db: Db; companyId: string; stamp: unknown; providerLeaseId: string;
|
||||
}): Promise<boolean> {
|
||||
const stamp = readNativeHarnessBackupStamp(input.stamp, input.providerLeaseId);
|
||||
if (!stamp) return false;
|
||||
const storage = durableNativeHarnessStorage();
|
||||
if (!storage) return verifyNativeHarnessBackupStamp(stamp, input.providerLeaseId);
|
||||
const durable = await verifyDurableNativeHarnessBackup({ db: input.db, storage, companyId: input.companyId,
|
||||
scopeDigest: stamp.sessionScopeSha256, manifestSha256: stamp.manifestSha256, signal: AbortSignal.timeout(300_000) });
|
||||
return durable || process.env.PAPERCLIP_EPHEMERAL_SERVER !== "true" && verifyNativeHarnessBackupStamp(stamp, input.providerLeaseId);
|
||||
}
|
||||
function reference(row: Row): Reference {
|
||||
const ref = row.reference as Reference;
|
||||
if (ref.schema !== SCHEMA || ref.id !== row.id || ref.companyId !== row.companyId || ref.scopeDigest !== row.scopeDigest
|
||||
|| !/^[a-f0-9]{64}$/.test(ref.scopeDigest) || !/^[a-f0-9]{64}$/.test(ref.sha256)
|
||||
|| ref.manifestSha256 !== row.manifestSha256 || !/^sha256:[a-f0-9]{64}$/.test(ref.manifestSha256)
|
||||
|| ref.objectKey !== `${row.companyId}/native-harness-backups/${row.scopeDigest}/${row.id}.pchn`
|
||||
|| !Number.isSafeInteger(ref.byteSize) || ref.byteSize <= 0 || ref.byteSize > 300 * 1024 * 1024
|
||||
|| !ref.material || typeof ref.material !== "object") fail("reference_invalid");
|
||||
return ref;
|
||||
}
|
||||
async function digest(file: string, signal?: AbortSignal) {
|
||||
const hash = createHash("sha256"); let byteSize = 0;
|
||||
for await (const bytes of createReadStream(file, { signal })) { hash.update(bytes); byteSize += bytes.length; }
|
||||
return { sha256: hash.digest("hex"), byteSize };
|
||||
}
|
||||
async function receive(storage: StorageService, ref: Reference, target?: string, signal?: AbortSignal) {
|
||||
const object = await storage.getObject(ref.companyId, ref.objectKey, { signal });
|
||||
const hash = createHash("sha256"); let byteSize = 0;
|
||||
const observed = new Transform({ transform(bytes, _encoding, callback) {
|
||||
hash.update(bytes); byteSize += bytes.length;
|
||||
callback(byteSize > ref.byteSize ? new Error("native_harness_storage_size_mismatch") : null, bytes);
|
||||
} });
|
||||
try {
|
||||
if (target) await pipeline(object.stream, observed, createWriteStream(target, { flags: "wx", mode: 0o600 }), { signal });
|
||||
else for await (const bytes of object.stream) {
|
||||
signal?.throwIfAborted(); hash.update(bytes); byteSize += bytes.length;
|
||||
if (byteSize > ref.byteSize) fail("size_mismatch");
|
||||
}
|
||||
if (byteSize !== ref.byteSize || hash.digest("hex") !== ref.sha256) fail("hash_mismatch");
|
||||
} finally { object.stream.destroy(); }
|
||||
}
|
||||
|
||||
export async function nativeHarnessBackupOwner(db: Db, input: { companyId: string; issueId: string; runId: string; bootId: string }) {
|
||||
const [row] = await db.select().from(nativeRunFinalizations).where(and(
|
||||
eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.issueId, input.issueId),
|
||||
eq(nativeRunFinalizations.runId, input.runId), eq(nativeRunFinalizations.controllerBootId, input.bootId)));
|
||||
if (!row?.leaseOwner) fail("owner_missing");
|
||||
return { ...input, authorityRunId: input.runId, leaseOwner: row!.leaseOwner!, generation: row!.controllerGeneration } satisfies NativeRunnerCheckpointOwner;
|
||||
}
|
||||
async function fence(tx: Parameters<Parameters<Db["transaction"]>[0]>[0], owner: NativeRunnerCheckpointOwner | OfflineOwner) {
|
||||
const [task] = await tx.select({ executionRunId: issues.executionRunId }).from(issues)
|
||||
.where(and(eq(issues.companyId, owner.companyId), eq(issues.id, owner.issueId))).for("share");
|
||||
const authorityRunId = "kind" in owner ? owner.runId : owner.authorityRunId;
|
||||
if (!task || !("kind" in owner) && task.executionRunId && task.executionRunId !== authorityRunId) fail("owner_superseded");
|
||||
if ("kind" in owner) {
|
||||
if (!owner.writersStopped) fail("offline_writers_required");
|
||||
const [active] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(offlineMigrationExecutionWhere()).limit(1);
|
||||
if (active) fail("offline_writers_required");
|
||||
const [run] = await tx.select({ finishedAt: heartbeatRuns.finishedAt, status: heartbeatRuns.status }).from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.id, owner.runId), eq(heartbeatRuns.companyId, owner.companyId),
|
||||
eq(heartbeatRuns.nativeIssueId, owner.issueId), eq(heartbeatRuns.runtimeMode, "native"))).for("share");
|
||||
const [coordinator] = await tx.select({ leaseOwner: nativeRunFinalizations.leaseOwner,
|
||||
leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt })
|
||||
.from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, owner.runId)).for("share");
|
||||
const [clock] = await tx.execute<{ now: Date }>(sql`select clock_timestamp() as now`);
|
||||
if (!run?.finishedAt || !["succeeded", "failed", "cancelled", "interrupted", "timed_out"].includes(run.status)
|
||||
|| !coordinator || coordinator.leaseOwner && (!coordinator.leaseExpiresAt || coordinator.leaseExpiresAt > new Date(clock.now))) {
|
||||
fail("offline_owner_unsettled");
|
||||
}
|
||||
return;
|
||||
}
|
||||
await assertNativeRunnerCheckpointOwner(tx, owner, false);
|
||||
}
|
||||
|
||||
export async function persistNativeHarnessBackup(input: {
|
||||
db: Db; storage: StorageService; owner: NativeRunnerCheckpointOwner | OfflineOwner;
|
||||
scopeDigest: string; root: string; signal?: AbortSignal;
|
||||
}): Promise<void> {
|
||||
if (!/^[a-f0-9]{64}$/.test(input.scopeDigest)) fail("scope_invalid");
|
||||
const manifestBytes = await readFile(join(input.root, "manifest.json"));
|
||||
const manifestSha256 = `sha256:${createHash("sha256").update(manifestBytes).digest("hex")}`;
|
||||
const binding = { companyId: input.owner.companyId, scopeDigest: input.scopeDigest, id: randomUUID() };
|
||||
const scratch = await mkdtemp(join(os.tmpdir(), "pc-harness-upload-"));
|
||||
const key = randomBytes(32);
|
||||
try {
|
||||
const archive = join(scratch, "archive"), encrypted = join(scratch, "encrypted");
|
||||
await captureNativeHarnessArchive(input.root, archive, input.signal);
|
||||
const plaintext = await digest(archive, input.signal), iv = randomBytes(12);
|
||||
const cipher = createCipheriv("aes-256-gcm", key, iv); cipher.setAAD(aad(binding));
|
||||
await pipeline(createReadStream(archive), cipher, createWriteStream(encrypted, { flags: "wx", mode: 0o600 }), { signal: input.signal });
|
||||
const ref: Reference = { schema: SCHEMA, ...binding, manifestSha256,
|
||||
objectKey: `${binding.companyId}/native-harness-backups/${binding.scopeDigest}/${binding.id}.pchn`,
|
||||
...await digest(encrypted, input.signal),
|
||||
material: createLocalEncryptedCipher("write").encrypt(JSON.stringify({ schema: SCHEMA, ...binding,
|
||||
key: key.toString("base64"), iv: iv.toString("base64"), tag: cipher.getAuthTag().toString("base64"), plaintext })) };
|
||||
await input.db.transaction(async tx => {
|
||||
await fence(tx, input.owner);
|
||||
await tx.insert(nativeHarnessBackups).values({ ...binding, issueId: input.owner.issueId, runId: input.owner.runId,
|
||||
manifestSha256, reference: ref });
|
||||
});
|
||||
await input.storage.putFile({ companyId: ref.companyId, namespace: "native-harness-backups", objectKey: ref.objectKey,
|
||||
originalFilename: null, contentType: "application/octet-stream", body: createReadStream(encrypted),
|
||||
byteSize: ref.byteSize, sha256: ref.sha256, signal: input.signal });
|
||||
await receive(input.storage, ref, undefined, input.signal);
|
||||
await input.db.transaction(async tx => {
|
||||
await fence(tx, input.owner);
|
||||
await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${`${ref.companyId}:${ref.scopeDigest}`}, 0))`);
|
||||
const [successor] = await tx.select({ id: nativeHarnessBackups.id }).from(nativeHarnessBackups)
|
||||
.innerJoin(heartbeatRuns, eq(heartbeatRuns.id, nativeHarnessBackups.runId)).where(and(scope(ref),
|
||||
eq(nativeHarnessBackups.status, "current"), sql`${nativeHarnessBackups.runId} <> ${input.owner.runId}::uuid`,
|
||||
sql`${heartbeatRuns.createdAt} >= (select created_at from heartbeat_runs where id = ${input.owner.runId}::uuid)`)).limit(1);
|
||||
if (successor) fail("owner_superseded");
|
||||
await fence(tx, input.owner);
|
||||
await tx.update(nativeHarnessBackups).set({ status: "superseded", updatedAt: new Date() })
|
||||
.where(and(scope(ref), eq(nativeHarnessBackups.status, "current")));
|
||||
await tx.update(nativeHarnessBackups).set({ status: "current", updatedAt: new Date() })
|
||||
.where(and(eq(nativeHarnessBackups.id, ref.id), eq(nativeHarnessBackups.status, "pending")));
|
||||
});
|
||||
} finally { key.fill(0); await removeNativeHarnessBackup(scratch); }
|
||||
}
|
||||
|
||||
export async function restoreDurableNativeHarnessBackup(input: {
|
||||
db: Db; storage: StorageService; companyId: string; scopeDigest: string;
|
||||
backupRoot: string; signal?: AbortSignal;
|
||||
}): Promise<boolean> {
|
||||
const [row] = await input.db.select().from(nativeHarnessBackups).where(and(scope(input), eq(nativeHarnessBackups.status, "current")));
|
||||
if (!row) return false;
|
||||
const ref = reference(row);
|
||||
const wrapped = JSON.parse(createLocalEncryptedCipher("read").decrypt(ref.material));
|
||||
if (wrapped.schema !== SCHEMA || wrapped.id !== ref.id || wrapped.companyId !== ref.companyId
|
||||
|| wrapped.scopeDigest !== ref.scopeDigest) fail("binding_mismatch");
|
||||
const current = join(input.backupRoot, "current");
|
||||
const localManifest = await readFile(join(current, "manifest.json")).catch((error: NodeJS.ErrnoException) => {
|
||||
if (error.code === "ENOENT") return null; throw error;
|
||||
});
|
||||
if (localManifest && `sha256:${createHash("sha256").update(localManifest).digest("hex")}` === ref.manifestSha256) return true;
|
||||
const scratch = await mkdtemp(join(os.tmpdir(), "pc-harness-restore-"));
|
||||
const key = Buffer.from(wrapped.key, "base64");
|
||||
try {
|
||||
if (key.length !== 32) fail("key_invalid");
|
||||
const encrypted = join(scratch, "encrypted"), archive = join(scratch, "archive");
|
||||
await receive(input.storage, ref, encrypted, input.signal);
|
||||
const decipher = createDecipheriv("aes-256-gcm", key, Buffer.from(wrapped.iv, "base64"));
|
||||
decipher.setAAD(aad(ref)); decipher.setAuthTag(Buffer.from(wrapped.tag, "base64"));
|
||||
await pipeline(createReadStream(encrypted), decipher, createWriteStream(archive, { flags: "wx", mode: 0o600 }), { signal: input.signal });
|
||||
const actual = await digest(archive, input.signal);
|
||||
if (actual.sha256 !== wrapped.plaintext?.sha256 || actual.byteSize !== wrapped.plaintext?.byteSize) fail("plaintext_mismatch");
|
||||
const restored = join(scratch, "restored");
|
||||
await restoreNativeHarnessArchive(archive, restored, input.signal);
|
||||
const manifest = await readFile(join(restored, "manifest.json"));
|
||||
if (`sha256:${createHash("sha256").update(manifest).digest("hex")}` !== ref.manifestSha256) fail("manifest_mismatch");
|
||||
const { mkdir, cp } = await import("node:fs/promises");
|
||||
await mkdir(input.backupRoot, { recursive: true, mode: 0o700 });
|
||||
const pending = join(input.backupRoot, `.durable-${randomUUID()}`);
|
||||
try {
|
||||
await cp(restored, pending, { recursive: true, preserveTimestamps: true, errorOnExist: true, force: false });
|
||||
await removeNativeHarnessBackup(current);
|
||||
await rename(pending, current);
|
||||
} finally { await removeNativeHarnessBackup(pending); }
|
||||
return true;
|
||||
} finally { key.fill(0); await removeNativeHarnessBackup(scratch); }
|
||||
}
|
||||
|
||||
export async function verifyDurableNativeHarnessBackup(input: {
|
||||
db: Db; storage: StorageService; companyId: string; scopeDigest: string; manifestSha256: string; signal?: AbortSignal;
|
||||
}): Promise<boolean> {
|
||||
const [row] = await input.db.select().from(nativeHarnessBackups).where(and(scope(input),
|
||||
eq(nativeHarnessBackups.manifestSha256, input.manifestSha256), inArray(nativeHarnessBackups.status, ["current", "superseded"]))).limit(1);
|
||||
if (!row) return false;
|
||||
await receive(input.storage, reference(row), undefined, input.signal);
|
||||
return true;
|
||||
}
|
||||
|
||||
export async function deleteRetiredNativeHarnessBackup(db: Db, storage: StorageService, id: string, signal?: AbortSignal) {
|
||||
const [row] = await db.select().from(nativeHarnessBackups).where(and(
|
||||
eq(nativeHarnessBackups.id, id), eq(nativeHarnessBackups.status, "retired")));
|
||||
if (!row) return false;
|
||||
const ref = reference(row);
|
||||
await storage.deleteObject(ref.companyId, ref.objectKey, { allVersions: true, signal });
|
||||
await db.delete(nativeHarnessBackups).where(and(eq(nativeHarnessBackups.id, id), eq(nativeHarnessBackups.status, "retired")));
|
||||
return true;
|
||||
}
|
||||
@@ -6,6 +6,8 @@ import { createLocalEncryptedCipher } from "../../secrets/local-encrypted-provid
|
||||
import { legacyControllerBootId } from "../legacy-controller-lease.js";
|
||||
|
||||
type Binding = { companyId: string; issueId: string; runId: string };
|
||||
export const NATIVE_CHECKPOINT_HEADER_INSPECTION_MAX_BYTES = 1024 * 1024;
|
||||
export const NATIVE_CHECKPOINT_INSPECTION_TOTAL_MAX_BYTES = 8 * 1024 * 1024;
|
||||
export type NativeRunnerCheckpointOwner = Binding & { bootId: string; leaseOwner: string; generation: number; authorityRunId: string };
|
||||
function encryptParts(binding: Binding, records: ReadonlyMap<string, string>) {
|
||||
if (!records.size) return [];
|
||||
@@ -16,6 +18,11 @@ function encryptParts(binding: Binding, records: ReadonlyMap<string, string>) {
|
||||
}));
|
||||
}
|
||||
|
||||
export async function insertMigratedNativeRunnerCheckpoint(tx: CheckpointTransaction, binding: Binding, checkpoint: DurablePrpCheckpoint) {
|
||||
if (checkpoint.runId !== binding.runId) throw new Error("native_runner_checkpoint_binding_mismatch");
|
||||
await tx.insert(nativeRunnerCheckpoints).values(encryptParts(binding, checkpoint.records));
|
||||
}
|
||||
|
||||
function decryptPart(cipher: ReturnType<typeof createLocalEncryptedCipher>, binding: Binding,
|
||||
part: string, material: Record<string, unknown>): string {
|
||||
const envelope = JSON.parse(cipher.decrypt(material));
|
||||
|
||||
@@ -4,6 +4,7 @@ import { providerFailureFromError } from "../../vendor/paperclip-runner/index.js
|
||||
import { nativeCancellationHasUnsettledOwner, notifyNativeCancellation, waitForNativeCancellation } from "./native-cancellation-delivery.js";
|
||||
import { readClaimedNativeStopRecovery } from "./native-cancellation-request.js";
|
||||
import { publishNativeHarnessBackup } from "./native-harness-backup-publication.js";
|
||||
import { durableNativeHarnessStorage, nativeHarnessBackupOwner, persistNativeHarnessBackup, restoreDurableNativeHarnessBackup } from "./native-harness-backup-storage.js";
|
||||
import { rehydratePinnedSkillAssets } from "./runtime-context.js";
|
||||
import { createNativeUsageDelivery } from "./native-usage-delivery.js";
|
||||
import { createNativeRecoveryUsageDelivery } from "./native-recovery-usage-delivery.js";
|
||||
@@ -5964,6 +5965,7 @@ function nativeSessionConfigDigest(
|
||||
execution: NativeExecutionInput,
|
||||
executionTargetKind: "local" | "remote" = "local",
|
||||
legacyProjectlessRunId?: string,
|
||||
recordedToolContractFingerprint?: string,
|
||||
): string {
|
||||
const executionLocation = {
|
||||
executionKind: "local_process",
|
||||
@@ -5999,13 +6001,25 @@ function nativeSessionConfigDigest(
|
||||
// changes so an older thread is rotated instead of falsely resuming
|
||||
// without newly required tools.
|
||||
nativeToolContractFingerprint:
|
||||
nativeToolContractFingerprintForTarget(executionTargetKind, execution.provider.kind),
|
||||
recordedToolContractFingerprint ?? nativeToolContractFingerprintForTarget(executionTargetKind, execution.provider.kind),
|
||||
runtimeContract: nativeRuntimeContractForProvider(execution.provider),
|
||||
}),
|
||||
)
|
||||
.digest("hex")}`;
|
||||
}
|
||||
|
||||
export function nativeControllerMigrationDescriptor(execution: NativeExecutionInput, recordedToolContractFingerprint: unknown) {
|
||||
if (typeof recordedToolContractFingerprint !== "string" || !/^sha256:[a-f0-9]{64}$/.test(recordedToolContractFingerprint)) {
|
||||
throw new Error("native_controller_migration_tool_contract_missing");
|
||||
}
|
||||
return {
|
||||
root: scopedRunnerdStateRoot(execution),
|
||||
supervisorPath: resolve(resolvePaperclipInstanceRoot(), "runtime", "paperclip-runner", "sessions",
|
||||
`${createHash("sha256").update(nativeSessionScopeKey(execution)).digest("hex")}.json`),
|
||||
configDigest: nativeSessionConfigDigest(execution, "remote", undefined, recordedToolContractFingerprint),
|
||||
};
|
||||
}
|
||||
|
||||
function idleWarmNativeSessionOwnerRunId(input: {
|
||||
execution: NativeExecutionInput;
|
||||
runnerExecutionTarget?: AdapterExecutionTarget | null;
|
||||
@@ -6027,6 +6041,7 @@ async function restorePerTurnRunnerdController(input: {
|
||||
if (input.restartRecovery || input.runnerExecutionTarget?.kind !== "remote" ||
|
||||
input.execution.session.lifecyclePolicy.mode !== "per_turn") return;
|
||||
const root = scopedRunnerdStateRoot(input.execution);
|
||||
await restoreDurableHarnessForExecution(input.db, input.execution, root);
|
||||
const identity = readRunnerdDurableIdentity(root);
|
||||
const reference = identity ?? readRunnerdDurableIdentity(root, true);
|
||||
if (!reference) {
|
||||
@@ -6091,6 +6106,7 @@ async function restoreWarmRunnerdController(input: {
|
||||
if (input.restartRecovery || input.runnerExecutionTarget?.kind !== "remote" ||
|
||||
input.execution.session.lifecyclePolicy.mode !== "warm") return;
|
||||
const root = scopedRunnerdStateRoot(input.execution);
|
||||
await restoreDurableHarnessForExecution(input.db, input.execution, root);
|
||||
const identity = readRunnerdDurableIdentity(root);
|
||||
const reference = identity ?? readRunnerdDurableIdentity(root, true);
|
||||
if (!idleWarmNativeSessionOwnerRunId(input) && (!reference || !existsSync(nativeSessionCheckpointPath(input.execution)))) {
|
||||
@@ -6599,6 +6615,14 @@ function harnessBackupRoot(root: string): string {
|
||||
return resolve(root, "failover-backups");
|
||||
}
|
||||
|
||||
async function restoreDurableHarnessForExecution(db: Db, execution: NativeExecutionInput, root: string) {
|
||||
const storage = durableNativeHarnessStorage();
|
||||
if (!storage) return;
|
||||
await restoreDurableNativeHarnessBackup({ db, storage, companyId: execution.binding.companyId,
|
||||
scopeDigest: createHash("sha256").update(nativeSessionScopeKey(execution)).digest("hex"),
|
||||
backupRoot: harnessBackupRoot(root), signal: AbortSignal.timeout(300_000) });
|
||||
}
|
||||
|
||||
function harnessBackupCandidates(root: string): string[] {
|
||||
const backupRoot = harnessBackupRoot(root);
|
||||
return [resolve(backupRoot, "current"), resolve(backupRoot, "previous")];
|
||||
@@ -14069,6 +14093,11 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
!remoteRunnerFilesystemRoot
|
||||
)
|
||||
return;
|
||||
const durableStorage = durableNativeHarnessStorage();
|
||||
if (durableStorage && !input.controllerPersistence) throw new Error("native_harness_storage_owner_missing");
|
||||
const backupOwner = durableStorage ? await nativeHarnessBackupOwner(input.db, { ...input.execution.binding,
|
||||
runId: sessionToolAuthorityEpochs.get(sessionScopeId)?.runId ?? input.execution.binding.runId,
|
||||
bootId: input.controllerPersistence!.bootId }) : null;
|
||||
// Transport release also runs when provider bootstrap failed. In that case
|
||||
// runnerd has no durable provider identity (and may not have created the
|
||||
// provider persistence directory at all), so attempting a failover backup
|
||||
@@ -14174,6 +14203,11 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
});
|
||||
renameSync(temporaryManifest, manifestPath);
|
||||
|
||||
if (durableStorage && backupOwner) {
|
||||
await persistNativeHarnessBackup({ db: input.db, storage: durableStorage, owner: backupOwner,
|
||||
scopeDigest: createHash("sha256").update(sessionScopeId).digest("hex"), root: pendingRoot,
|
||||
signal: AbortSignal.timeout(300_000) });
|
||||
}
|
||||
await publishNativeHarnessBackup({
|
||||
db: input.db, companyId: input.execution.binding.companyId,
|
||||
scopeDigest: createHash("sha256").update(sessionScopeId).digest("hex"), backupRoot, pendingRoot,
|
||||
|
||||
@@ -225,7 +225,7 @@ type NativeSessionResumeSelection = {
|
||||
agentId: string;
|
||||
issueId: string;
|
||||
normalizedSessionId: string;
|
||||
currentRunId: string;
|
||||
currentRunId?: string;
|
||||
beforeCreatedAt: Date;
|
||||
/** Recovery discovery must retain an ineligible newest progress barrier. */
|
||||
includeProgressBarrier?: boolean;
|
||||
@@ -306,7 +306,7 @@ async function selectNativeSessionResumeBarrier(db: Db, input: NativeSessionResu
|
||||
.from(heartbeatRuns)
|
||||
.where(
|
||||
and(
|
||||
ne(heartbeatRuns.id, input.currentRunId),
|
||||
input.currentRunId ? ne(heartbeatRuns.id, input.currentRunId) : undefined,
|
||||
eq(heartbeatRuns.companyId, input.companyId),
|
||||
eq(heartbeatRuns.agentId, input.agentId),
|
||||
eq(heartbeatRuns.nativeIssueId, input.issueId),
|
||||
|
||||
@@ -73,28 +73,22 @@ export function shouldUseRemoteWorkspaces(input: {
|
||||
return configuredRemoteWorkspacePersistence() !== null;
|
||||
}
|
||||
|
||||
/** Deployment profile admission belongs beside workspace authority. A warm
|
||||
* remote controller may replace its local projection; host-owned execution
|
||||
* cannot make that promise. External conversation providers own no task disk. */
|
||||
export function assertEphemeralWorkspaceExecution(input: {
|
||||
externalConversation: boolean; adapterType: string;
|
||||
pluginTaskWorkspace: boolean; remoteAuthority: boolean;
|
||||
}) {
|
||||
if (process.env.PAPERCLIP_EPHEMERAL_SERVER !== "true" || input.externalConversation) return;
|
||||
// Runtime selection has not promoted a fresh run yet. Validate storage here;
|
||||
// the runtime selector separately admits only its resolved warm native policy.
|
||||
if (input.adapterType !== "paperclip_runner"
|
||||
|| (!input.pluginTaskWorkspace && !input.remoteAuthority)) {
|
||||
throw unprocessable("Volume-free servers require warm Paperclip Runner execution with remote workspace storage.",
|
||||
throw unprocessable("Volume-free servers require native Paperclip Runner execution with remote workspace storage.",
|
||||
{ code: "ephemeral_execution_configuration_required" });
|
||||
}
|
||||
}
|
||||
|
||||
/** The runtime selector calls this after environment, credentials and saved
|
||||
* execution policy have resolved. Raw agent settings are not lifecycle authority. */
|
||||
export function assertEphemeralNativeLifecycle(mode: "warm" | "per_turn" | undefined) {
|
||||
if (process.env.PAPERCLIP_EPHEMERAL_SERVER === "true" && mode !== "warm") {
|
||||
throw unprocessable("Volume-free servers require warm Paperclip Runner execution with remote workspace storage.",
|
||||
export function assertEphemeralNativeRuntime(native: boolean) {
|
||||
if (process.env.PAPERCLIP_EPHEMERAL_SERVER === "true" && !native) {
|
||||
throw unprocessable("Volume-free servers require native Paperclip Runner execution with remote workspace storage.",
|
||||
{ code: "ephemeral_execution_configuration_required" });
|
||||
}
|
||||
}
|
||||
|
||||
+1
@@ -94,6 +94,7 @@ export const RUNNER_TOOL_SCOPE_TOKEN_HEADER = runner.RUNNER_TOOL_SCOPE_TOKEN_HEA
|
||||
export const DurablePrpControlPlane = runner.DurablePrpControlPlane;
|
||||
export const providerFailureFromError = runner.providerFailureFromError;
|
||||
export const decodeDurablePrpCheckpoint = runner.decodeDurablePrpCheckpoint;
|
||||
export const captureLegacyDurablePrpCheckpoint = runner.captureLegacyDurablePrpCheckpoint;
|
||||
export const inspectDurablePrpCheckpointHeader = runner.inspectDurablePrpCheckpointHeader;
|
||||
export const RUNNER_PRP_PROTOCOL_MIN = runner.RUNNER_PRP_PROTOCOL_MIN;
|
||||
export const RUNNER_PRP_PROTOCOL_MAX = runner.RUNNER_PRP_PROTOCOL_MAX;
|
||||
|
||||
Reference in new issue
Block a user