mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
Preserve settled failed controller state during online migration
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
961455b9a0
commit
5a6032c14b
7 files changed
+233
-13
No files matched your search
@@ -343,6 +343,28 @@ Unrelated runs remain active. Existing controller checkpoint identity, artifact
|
||||
provenance, configuration and snapshot checks still apply. The `paperclip.native-controller-prestage.v1` counts describe this phase; neither successful prestaging
|
||||
nor zero deferred controller sources authorizes volume detachment.
|
||||
|
||||
The same finite controller-only entry also preserves settled failed sources.
|
||||
A committed failed result may import its exact checkpoint under the normal
|
||||
owner checks without changing the failed outcome. A terminal failed controller
|
||||
that cannot earn executable recovery authority instead gets an encrypted
|
||||
evidence archive through the existing native harness object owner.
|
||||
|
||||
Online archival verifies the exact retained session pointer, company, agent,
|
||||
task, admitted source binding, terminal run and released coordinator. It rejects
|
||||
related queued/running work, checkout, expired-but-retained ownership and a busy
|
||||
physical finalizer. Source/profile/coordinator fingerprints and original file
|
||||
bytes are checked at publication, including after the upload. Pending object
|
||||
records commit before upload; a raced source cannot publish a current archive.
|
||||
The operation leaves original files, failed status, leases and recovery history
|
||||
unchanged. The `archivedFailed` count records preservation, never successful
|
||||
execution or permission to take over an agent.
|
||||
|
||||
The live Cloud path uses online preparation and static ephemeral arming on the
|
||||
retained volume. Replacement startup must recheck the original bytes after the
|
||||
old deployment stops. That final check, not an archive copy or process-local
|
||||
busy map, establishes the cutover boundary. The offline fallback below is for
|
||||
an independently quiet instance; it is not the live staging delivery path.
|
||||
|
||||
The final offline migration still stops all writers, verifies every durable
|
||||
object, imports deferred sources and all other data, and exports only its normal
|
||||
complete eligibility receipt. Qualification starts from the saved pristine
|
||||
|
||||
@@ -0,0 +1,89 @@
|
||||
import { and, eq, inArray, or, sql } from "drizzle-orm";
|
||||
import { createHash } from "node:crypto";
|
||||
import { agentTaskSessions, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db";
|
||||
import { nativeSha256 } from "./canonical.js";
|
||||
import { findNativeSessionResumeRun } from "./native-session-resume.js";
|
||||
import { readLegacyNativeControllerArchiveSource } from "./native-controller-migration.js";
|
||||
import { tryLockNativeWorkspaceFinalization } from "./native-workspace-finalization-ownership.js";
|
||||
import { archiveFailedLegacyNativeController } from "./native-controller-archive.js";
|
||||
import type { StorageService } from "../../storage/types.js";
|
||||
|
||||
export type SettledControllerArchiveOwner = {
|
||||
kind: "settled_controller_archive";
|
||||
companyId: string; issueId: string; agentId: string; runId: string; sessionId: string;
|
||||
selectedSessionId: string; selectedLastRunId: string;
|
||||
profileDigest: string; coordinatorDigest: string; controllerDigest: string; supervisorDigest: string | null;
|
||||
};
|
||||
const fileDigest = (value: string) => createHash("sha256").update(value).digest("hex");
|
||||
|
||||
export function nativeControllerArchiveProfileDigest(value: unknown) {
|
||||
const profile = value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : {};
|
||||
return nativeSha256({ nativeExecutionInput: profile.nativeExecutionInput ?? null, nativeToolContractFingerprint: profile.nativeToolContractFingerprint ?? null,
|
||||
sessionCheckpoint: profile.sessionCheckpoint ?? null, nativeControllerRecovery: profile.nativeControllerRecovery ?? null });
|
||||
}
|
||||
|
||||
export function nativeControllerArchiveCoordinatorDigest(value: {
|
||||
phase: string; resultId: string | null; controllerBootId: string | null; controllerGeneration: number; recoveryState: unknown;
|
||||
}) {
|
||||
return nativeSha256({ phase: value.phase, resultId: value.resultId, controllerBootId: value.controllerBootId,
|
||||
controllerGeneration: value.controllerGeneration, recoveryState: value.recoveryState });
|
||||
}
|
||||
|
||||
export async function assertSettledControllerArchiveSource(db: Db, owner: SettledControllerArchiveOwner) {
|
||||
function fail(): never { throw new Error("native_controller_archive_source_unsettled"); }
|
||||
await db.execute(sql`set local lock_timeout = '1s'`);
|
||||
if (!await tryLockNativeWorkspaceFinalization(db, owner)) fail();
|
||||
const [task] = await db.select().from(issues).where(and(eq(issues.companyId, owner.companyId), eq(issues.id, owner.issueId))).for("update");
|
||||
if (!task || task.executionRunId || task.checkoutRunId) fail();
|
||||
const [coordinator] = await db.select().from(nativeRunFinalizations).where(and(
|
||||
eq(nativeRunFinalizations.companyId, owner.companyId), eq(nativeRunFinalizations.issueId, owner.issueId),
|
||||
eq(nativeRunFinalizations.runId, owner.runId))).for("update");
|
||||
if (!coordinator || coordinator.leaseOwner || coordinator.nextAttemptAt
|
||||
|| !(coordinator.phase === "terminal_failure" || coordinator.phase === "committed" && coordinator.resultId)
|
||||
|| nativeControllerArchiveCoordinatorDigest(coordinator) !== owner.coordinatorDigest) fail();
|
||||
const [selection] = await db.select().from(agentTaskSessions).where(and(eq(agentTaskSessions.id, owner.selectedSessionId),
|
||||
eq(agentTaskSessions.companyId, owner.companyId), eq(agentTaskSessions.agentId, owner.agentId),
|
||||
eq(agentTaskSessions.lastRunId, owner.selectedLastRunId))).for("share");
|
||||
const [selectedRun] = await db.select({ createdAt: heartbeatRuns.createdAt }).from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.id, owner.selectedLastRunId), eq(heartbeatRuns.companyId, owner.companyId), eq(heartbeatRuns.agentId, owner.agentId),
|
||||
eq(heartbeatRuns.nativeIssueId, owner.issueId), eq(heartbeatRuns.nativeSessionId, owner.sessionId)));
|
||||
const [run] = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, owner.runId),
|
||||
eq(heartbeatRuns.companyId, owner.companyId), eq(heartbeatRuns.agentId, owner.agentId), eq(heartbeatRuns.nativeIssueId, owner.issueId),
|
||||
eq(heartbeatRuns.nativeSessionId, owner.sessionId), eq(heartbeatRuns.runtimeMode, "native"))).for("share");
|
||||
if (!selection || !selectedRun || !run?.finishedAt || run.status !== "failed"
|
||||
|| nativeControllerArchiveProfileDigest(run.runnerProfileJson) !== owner.profileDigest) fail();
|
||||
const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, owner.companyId),
|
||||
inArray(heartbeatRuns.status, ["queued", "running"]), or(eq(heartbeatRuns.nativeIssueId, owner.issueId),
|
||||
eq(heartbeatRuns.issueId, owner.issueId), eq(heartbeatRuns.nativeSessionId, owner.sessionId)))).limit(1);
|
||||
if (active) fail();
|
||||
const selected = await findNativeSessionResumeRun(db, { companyId: owner.companyId, agentId: owner.agentId, issueId: owner.issueId,
|
||||
normalizedSessionId: owner.sessionId, beforeCreatedAt: selectedRun.createdAt, includeProgressBarrier: true });
|
||||
if (!selected || selected.id !== owner.runId || nativeControllerArchiveProfileDigest(selected.runnerProfileJson) !== owner.profileDigest) fail();
|
||||
const material = await readLegacyNativeControllerArchiveSource(selected);
|
||||
if (material.controller === undefined || fileDigest(material.controller) !== owner.controllerDigest
|
||||
|| (material.supervisor === undefined ? null : fileDigest(material.supervisor)) !== owner.supervisorDigest) fail();
|
||||
}
|
||||
|
||||
export async function archiveSettledFailedNativeController(db: Db, input: {
|
||||
selected: { id: string; lastRunId: string; companyId: string; agentId: string; issueId: string; sessionId: string };
|
||||
source: NonNullable<Awaited<ReturnType<typeof findNativeSessionResumeRun>>>;
|
||||
storage: StorageService; signal: AbortSignal;
|
||||
}) {
|
||||
input.signal.throwIfAborted();
|
||||
const { selected, source } = input;
|
||||
const material = await readLegacyNativeControllerArchiveSource(source);
|
||||
if (material.controller === undefined) return "source_already_absent" as const;
|
||||
const [coordinator] = await db.select().from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, selected.companyId),
|
||||
eq(nativeRunFinalizations.issueId, selected.issueId), eq(nativeRunFinalizations.runId, source.id)));
|
||||
if (!coordinator) throw new Error("native_controller_archive_source_unsettled");
|
||||
const owner: SettledControllerArchiveOwner = { kind: "settled_controller_archive", companyId: selected.companyId,
|
||||
issueId: selected.issueId, agentId: selected.agentId, runId: source.id, sessionId: selected.sessionId,
|
||||
selectedSessionId: selected.id, selectedLastRunId: selected.lastRunId,
|
||||
profileDigest: nativeControllerArchiveProfileDigest(source.runnerProfileJson), coordinatorDigest: nativeControllerArchiveCoordinatorDigest(coordinator),
|
||||
controllerDigest: fileDigest(material.controller), supervisorDigest: material.supervisor === undefined ? null : fileDigest(material.supervisor) };
|
||||
await db.transaction(tx => assertSettledControllerArchiveSource(tx as unknown as Db, owner));
|
||||
input.signal.throwIfAborted();
|
||||
await archiveFailedLegacyNativeController(db, { companyId: selected.companyId, issueId: selected.issueId, runId: source.id,
|
||||
controller: material.controller, supervisor: material.supervisor, storage: input.storage, onlineOwner: owner, signal: input.signal });
|
||||
return "archived_failed" as const;
|
||||
}
|
||||
@@ -6,6 +6,7 @@ import type { Db } from "@paperclipai/db";
|
||||
import type { StorageService } from "../../storage/types.js";
|
||||
import { removeNativeHarnessBackup } from "./native-harness-backup-cleanup.js";
|
||||
import { persistNativeHarnessBackup, verifyDurableNativeHarnessBackup } from "./native-harness-backup-storage.js";
|
||||
import type { SettledControllerArchiveOwner } from "./native-controller-archive-source.js";
|
||||
|
||||
type Source = { companyId: string; issueId: string; runId: string; controller: string; supervisor?: string };
|
||||
const sha256 = (value: string) => createHash("sha256").update(value).digest("hex");
|
||||
@@ -24,8 +25,14 @@ export async function hasArchivedLegacyNativeController(db: Db, input: Source &
|
||||
}
|
||||
|
||||
export async function archiveFailedLegacyNativeController(db: Db, input: Source & {
|
||||
storage: StorageService; writersStopped: true; signal: AbortSignal;
|
||||
storage: StorageService; writersStopped?: true; onlineOwner?: SettledControllerArchiveOwner; signal: AbortSignal;
|
||||
}) {
|
||||
if (input.onlineOwner ? input.writersStopped || input.onlineOwner.kind !== "settled_controller_archive"
|
||||
|| input.onlineOwner.companyId !== input.companyId || input.onlineOwner.issueId !== input.issueId
|
||||
|| input.onlineOwner.runId !== input.runId || input.onlineOwner.controllerDigest !== sha256(input.controller)
|
||||
|| input.onlineOwner.supervisorDigest !== (input.supervisor === undefined ? null : sha256(input.supervisor)) : !input.writersStopped) {
|
||||
throw new Error("native_controller_archive_authority_invalid");
|
||||
}
|
||||
if (await hasArchivedLegacyNativeController(db, input)) return;
|
||||
const descriptor = legacyNativeControllerArchive(input);
|
||||
const root = await mkdtemp(join(tmpdir(), "pc-legacy-controller-archive-"));
|
||||
@@ -34,7 +41,7 @@ export async function archiveFailedLegacyNativeController(db: Db, input: Source
|
||||
await writeFile(join(root, "control-plane-state.json"), input.controller, { mode: 0o600 });
|
||||
if (input.supervisor !== undefined) await writeFile(join(root, "supervisor.json"), input.supervisor, { mode: 0o600 });
|
||||
await persistNativeHarnessBackup({ db, storage: input.storage, root, scopeDigest: descriptor.scopeDigest, signal: input.signal,
|
||||
owner: { kind: "offline_migration", writersStopped: input.writersStopped, terminalFailureOnly: true,
|
||||
owner: input.onlineOwner ?? { kind: "offline_migration", writersStopped: true, terminalFailureOnly: true,
|
||||
companyId: input.companyId, issueId: input.issueId, runId: input.runId } });
|
||||
} finally { await removeNativeHarnessBackup(root); }
|
||||
}
|
||||
@@ -22,7 +22,8 @@ import { createStorageService } from "../../storage/service.js";
|
||||
import { legacyNativeControllerArchive } from "./native-controller-archive.js";
|
||||
import { restoreDurableNativeHarnessBackup } from "./native-harness-backup-storage.js";
|
||||
|
||||
describeEmbeddedPostgres.each(["online_success", "online_owned", "online_same_source", "online_busy", "online_changed", "online_evidence", "import", "archive", "success", "batch", "durable_failed",
|
||||
describeEmbeddedPostgres.each(["archive_online", "archive_online_owned", "archive_online_changed", "archive_online_commit", "archive_online_active", "archive_online_foreign",
|
||||
"online_success", "online_owned", "online_same_source", "online_busy", "online_changed", "online_evidence", "import", "archive", "success", "batch", "durable_failed",
|
||||
"stale_confirmed", "stale_foreign", "stale_changed", "stale_queued", "stale_overlap", "stale_success", "stale_busy", "stale_durable", "stale_durable_absent",
|
||||
"stale_durable_runner", "stale_durable_session", "stale_durable_snapshot",
|
||||
"controller_confirmed", "controller_foreign", "controller_thread", "controller_runner", "controller_success", "controller_busy", "controller_invalid", "controller_durable",
|
||||
@@ -90,6 +91,77 @@ describeEmbeddedPostgres.each(["online_success", "online_owned", "online_same_so
|
||||
taskKey: f.issueId, lastRunId: f.runId }).returning();
|
||||
return { ...f, companyId, agentId, paperclipHome, runtimeRoot, paths, file, source, profile, session, queuedId, identity, snapshot };
|
||||
}
|
||||
if (scenario.startsWith("archive_online")) it("preserves a failed source online without changing its execution or unrelated work", async () => {
|
||||
const f = await fixture();
|
||||
await f.db.update(issues).set({ executionRunId: null, checkoutRunId: null }).where(eq(issues.id, f.issueId));
|
||||
await f.db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, f.queuedId));
|
||||
if (scenario === "archive_online_commit") {
|
||||
const [run] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
const [result] = await f.db.insert(nativeRunResults).values({ companyId: f.companyId, issueId: f.issueId, runId: f.runId,
|
||||
completionContractId: run.completionContractId!, serverFingerprint: "failed-committed-migration", schemaStatus: "accepted",
|
||||
resultJson: {}, canonicalSha256: "failed-committed-migration" }).returning();
|
||||
await f.db.update(nativeRunFinalizations).set({ phase: "committed", resultId: result.id, leaseOwner: null,
|
||||
leaseExpiresAt: null, recoveryState: null }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
} else {
|
||||
await rm(join(f.paths.root, "provider-pack-authority.json"));
|
||||
await f.db.update(nativeRunFinalizations).set({ recoveryState: "blocked", failureCode: "native_provider_terminal_failed" })
|
||||
.where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
}
|
||||
if (scenario === "archive_online_owned") await f.db.update(nativeRunFinalizations).set({
|
||||
leaseOwner: "expired-is-still-owned", leaseExpiresAt: new Date(0),
|
||||
}).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
if (scenario === "archive_online_foreign") await f.db.update(heartbeatRuns).set({ runnerProfileJson: { ...f.profile,
|
||||
nativeExecutionInput: { ...f.profile.nativeExecutionInput, binding: { ...f.profile.nativeExecutionInput.binding, companyId: randomUUID() } } } })
|
||||
.where(eq(heartbeatRuns.id, f.runId));
|
||||
const unrelated = await environment.seedRun(randomUUID());
|
||||
const [unrelatedBefore] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, unrelated.runId));
|
||||
const [runBefore] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
const [ownerBefore] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
const storage = createStorageService(createLocalDiskStorageProvider(join(f.runtimeRoot, "durable-objects")));
|
||||
const originalPut = storage.putFile.bind(storage);
|
||||
const put = vi.spyOn(storage, "putFile").mockImplementation(async input => {
|
||||
const result = await originalPut(input);
|
||||
if (scenario === "archive_online_changed") await writeFile(f.file, f.source + "\n", { mode: 0o600 });
|
||||
if (scenario === "archive_online_active") await f.db.insert(heartbeatRuns).values({ id: randomUUID(), companyId: f.companyId,
|
||||
agentId: f.agentId, issueId: f.issueId, nativeIssueId: f.issueId, nativeSessionId: f.native.normalizedSessionId,
|
||||
status: "queued", runtimeMode: "native" });
|
||||
return result;
|
||||
});
|
||||
const counts = await prestageNativeControllers(f.db, { storage, signal: AbortSignal.timeout(30_000) });
|
||||
if (["archive_online_owned", "archive_online_changed", "archive_online_active", "archive_online_foreign"].includes(scenario)) {
|
||||
expect(counts.archivedFailed).toBe(0);
|
||||
expect(counts.deferred[scenario === "archive_online_foreign" ? "artifact_binding" : "archive_source_unsettled"]).toBe(1);
|
||||
if (["archive_online_owned", "archive_online_changed"].includes(scenario)) expect(await hasLocalNativeControllerObligations(f.db, storage)).toBe(true);
|
||||
const backups = await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.runId, f.runId));
|
||||
expect(backups.some(backup => backup.status === "current")).toBe(false);
|
||||
if (scenario === "archive_online_owned") expect(put).not.toHaveBeenCalled();
|
||||
if (scenario === "archive_online_changed") expect(put).toHaveBeenCalledOnce();
|
||||
if (scenario === "archive_online_active") {
|
||||
expect(put).toHaveBeenCalledOnce();
|
||||
expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.issueId, f.issueId))).some(run => run.status === "queued")).toBe(true);
|
||||
}
|
||||
if (scenario === "archive_online_foreign") expect(put).not.toHaveBeenCalled();
|
||||
if (scenario === "archive_online_foreign") await expect(hasLocalNativeControllerObligations(f.db, storage))
|
||||
.rejects.toThrow("native_controller_migration_artifact_binding");
|
||||
} else if (scenario === "archive_online_commit") {
|
||||
expect(counts.imported).toBe(1);
|
||||
expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0].status).toBe("failed");
|
||||
expect(await loadNativeRunnerCheckpoint(f.db, { companyId: f.companyId, issueId: f.issueId, runId: f.runId })).not.toBeNull();
|
||||
} else {
|
||||
expect(counts.archivedFailed).toBe(1);
|
||||
expect(await hasLocalNativeControllerObligations(f.db, storage)).toBe(false);
|
||||
expect((await prestageNativeControllers(f.db, { storage, signal: AbortSignal.timeout(30_000) })).archivedFailed).toBe(1);
|
||||
expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]).toEqual(runBefore);
|
||||
const descriptor = legacyNativeControllerArchive({ companyId: f.companyId, issueId: f.issueId, runId: f.runId, controller: f.source });
|
||||
const replacement = join(f.runtimeRoot, "online-archive-inspection");
|
||||
await rm(f.paperclipHome, { recursive: true, force: true });
|
||||
expect(await restoreDurableNativeHarnessBackup({ db: f.db, storage, companyId: f.companyId,
|
||||
scopeDigest: descriptor.scopeDigest, backupRoot: replacement })).toBe(true);
|
||||
expect(await readFile(join(replacement, "current", "control-plane-state.json"), "utf8")).toBe(f.source);
|
||||
}
|
||||
expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toEqual(ownerBefore);
|
||||
expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, unrelated.runId)))[0]).toEqual(unrelatedBefore);
|
||||
});
|
||||
if (scenario.startsWith("online_")) it("prestages only a settled legacy controller while unrelated work continues", async () => {
|
||||
const f = await fixture();
|
||||
const settle = async (f: Awaited<ReturnType<typeof fixture>>) => {
|
||||
|
||||
@@ -17,8 +17,9 @@ import type { StorageService } from "../../storage/types.js";
|
||||
import { durableNativeHarnessStorage } from "./native-harness-backup-storage.js";
|
||||
import { archiveFailedLegacyNativeController, hasArchivedLegacyNativeController } from "./native-controller-archive.js";
|
||||
import { redactRunnerApiPayloads } from "./runner-api-log-redaction.js";
|
||||
import { archiveSettledFailedNativeController } from "./native-controller-archive-source.js";
|
||||
|
||||
type Selected = { id: string; companyId: string; agentId: string; issueId: string; sessionId: string; createdAt: Date };
|
||||
type Selected = { id: string; companyId: string; agentId: string; issueId: string; sessionId: string; createdAt: Date; lastRunId: string };
|
||||
|
||||
function record(value: unknown): Record<string, unknown> {
|
||||
return value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : {};
|
||||
@@ -79,7 +80,7 @@ async function* retainedSources(db: Db, signal?: AbortSignal) {
|
||||
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"
|
||||
r.native_issue_id as "issueId", r.native_session_id as "sessionId", r.created_at as "createdAt", r.id as "lastRunId"
|
||||
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
|
||||
@@ -110,6 +111,9 @@ async function readPrivateSource(file: string, maxBytes: number): Promise<string
|
||||
|
||||
async function localPaths(source: NonNullable<Awaited<ReturnType<typeof findNativeSessionResumeRun>>>) {
|
||||
const execution = parseNativeExecutionInput(source.runnerProfileJson.nativeExecutionInput);
|
||||
if (execution.binding.companyId !== source.companyId || execution.binding.agentId !== source.agentId
|
||||
|| execution.binding.issueId !== source.nativeIssueId || execution.binding.runId !== source.id
|
||||
|| execution.session.normalizedSessionId !== source.nativeSessionId) throw new Error("native_controller_migration_artifact_binding");
|
||||
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;
|
||||
@@ -123,6 +127,14 @@ async function localPaths(source: NonNullable<Awaited<ReturnType<typeof findNati
|
||||
return { ...descriptor, execution, controlFile: join(control, "control-plane-state.json") };
|
||||
}
|
||||
|
||||
export async function readLegacyNativeControllerArchiveSource(source: NonNullable<Awaited<ReturnType<typeof findNativeSessionResumeRun>>>) {
|
||||
const paths = await localPaths(source);
|
||||
const controller = await readPrivateSource(paths.controlFile, 256 * 1024 * 1024);
|
||||
const supervisor = paths.execution.session.lifecyclePolicy.mode === "warm"
|
||||
? await readPrivateSource(paths.supervisorPath, 4 * 1024 * 1024) : undefined;
|
||||
return { paths, controller, supervisor };
|
||||
}
|
||||
|
||||
function preservedFailedCheckpoint(source: NonNullable<Awaited<ReturnType<typeof findNativeSessionResumeRun>>>,
|
||||
sessionId: string, checkpoint: Awaited<ReturnType<typeof loadNativeRunnerCheckpoint>>) {
|
||||
return source.status === "failed" && source.runnerProfileJson.sessionCheckpoint == null
|
||||
@@ -264,6 +276,7 @@ function onlineControllerSourceDeferral(error: unknown): string | null {
|
||||
native_controller_migration_unsafe_source: "unsafe_source",
|
||||
native_controller_migration_artifact_binding: "artifact_binding",
|
||||
native_controller_migration_artifact_unavailable: "artifact_unavailable",
|
||||
native_controller_archive_source_unsettled: "archive_source_unsettled",
|
||||
};
|
||||
if (reasons[error.message]) return reasons[error.message];
|
||||
if (error.message.startsWith("runner_remote_provider_artifact_incompatible:")) return "artifact_incompatible";
|
||||
@@ -272,13 +285,13 @@ function onlineControllerSourceDeferral(error: unknown): string | null {
|
||||
return null;
|
||||
}
|
||||
|
||||
export async function prestageNativeControllers(db: Db, input: { signal: AbortSignal }) {
|
||||
const counts = { inspected: 0, imported: 0, unchanged: 0, sourceAlreadyAbsent: 0, deferred: {} as Record<string, number> };
|
||||
export async function prestageNativeControllers(db: Db, input: { signal: AbortSignal; storage?: StorageService }) {
|
||||
const counts = { inspected: 0, imported: 0, unchanged: 0, archivedFailed: 0, sourceAlreadyAbsent: 0, deferred: {} as Record<string, number> };
|
||||
const defer = (reason: string) => { counts.deferred[reason] = (counts.deferred[reason] ?? 0) + 1; };
|
||||
for await (const { selected, source } of retainedSources(db, input.signal)) {
|
||||
counts.inspected++;
|
||||
input.signal.throwIfAborted();
|
||||
const outcome = await db.transaction(async tx => {
|
||||
let outcome = await db.transaction(async tx => {
|
||||
await tx.execute(sql`set local lock_timeout = '1s'`);
|
||||
const [task] = await tx.select().from(issues).where(and(eq(issues.companyId, selected.companyId),
|
||||
eq(issues.id, selected.issueId))).for("update");
|
||||
@@ -290,7 +303,7 @@ export async function prestageNativeControllers(db: Db, input: { signal: AbortSi
|
||||
const [current] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, source.id),
|
||||
eq(heartbeatRuns.companyId, selected.companyId), eq(heartbeatRuns.agentId, selected.agentId),
|
||||
eq(heartbeatRuns.nativeIssueId, selected.issueId), eq(heartbeatRuns.nativeSessionId, selected.sessionId))).for("update");
|
||||
if (current?.status !== "succeeded" || !current.finishedAt) return "source_not_successful";
|
||||
if (!current || !["succeeded", "failed"].includes(current.status) || !current.finishedAt) return "source_not_successful";
|
||||
const [active] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, selected.companyId), inArray(heartbeatRuns.status, ["queued", "running"]),
|
||||
or(eq(heartbeatRuns.nativeIssueId, selected.issueId), eq(heartbeatRuns.issueId, selected.issueId),
|
||||
@@ -331,7 +344,18 @@ export async function prestageNativeControllers(db: Db, input: { signal: AbortSi
|
||||
if (error?.code === "55P03" || error?.cause?.code === "55P03") return "source_busy";
|
||||
throw error;
|
||||
});
|
||||
if (outcome === "imported" || outcome === "unchanged") counts[outcome]++;
|
||||
if (source.status === "failed" && ["owner_not_released", "artifact_unproven", "checkpoint_binding", "source_unproven", "artifact_binding"].includes(outcome)) {
|
||||
const storage = input.storage ?? durableNativeHarnessStorage();
|
||||
if (storage) outcome = await archiveSettledFailedNativeController(db, { selected, source, storage, signal: input.signal }).catch(error => {
|
||||
input.signal.throwIfAborted();
|
||||
const reason = onlineControllerSourceDeferral(error);
|
||||
if (reason) return reason;
|
||||
if (error?.code === "55P03" || error?.cause?.code === "55P03") return "source_busy";
|
||||
throw error;
|
||||
});
|
||||
}
|
||||
if (outcome === "archived_failed") counts.archivedFailed++;
|
||||
else if (outcome === "imported" || outcome === "unchanged") counts[outcome]++;
|
||||
else if (outcome === "source_already_absent") counts.sourceAlreadyAbsent++;
|
||||
else defer(outcome);
|
||||
}
|
||||
|
||||
@@ -266,7 +266,8 @@ export async function importNativeControllerRecovery(db: Db, input: {
|
||||
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" || input.offline && ["failed", "cancelled", "interrupted", "timed_out"].includes(run.status)) || !run.finishedAt
|
||||
if (run?.runtimeMode !== "native" || !(run.status === "succeeded" || input.legacyCheckpoint && run.status === "failed"
|
||||
|| 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
|
||||
|| input.supervisor !== undefined && nativeSha256(snapshot) !== nativeSha256(redactRunnerApiPayloads(supervisor.snapshot))) {
|
||||
|
||||
@@ -22,6 +22,7 @@ import { readNativeHarnessBackupSource } from "./native-harness-backup-source.js
|
||||
import { pinNativeProviderArtifactsForBackup, releaseRetiredNativeProviderArtifacts, restoreNativeProviderArtifacts, retainNativeProviderArtifacts,
|
||||
type NativeProviderArtifactReceipt, type NativeProviderArtifactSource } from "./native-provider-artifacts.js";
|
||||
import { canonicalProviderPackJson } from "./remote-provider-pack-verification.js";
|
||||
import type { SettledControllerArchiveOwner } from "./native-controller-archive-source.js";
|
||||
|
||||
const SCHEMA = "paperclip.native-harness-object.v1";
|
||||
type Binding = { companyId: string; scopeDigest: string; id: string };
|
||||
@@ -96,9 +97,13 @@ export async function nativeHarnessBackupOwner(db: Db, input: { companyId: strin
|
||||
if (!row || !row.leaseOwner && (row.phase !== "committed" || !row.resultId)) fail("owner_missing");
|
||||
return { ...input, authorityRunId: input.runId, leaseOwner: row!.leaseOwner, generation: row!.controllerGeneration } satisfies NativeRunnerCheckpointFenceOwner;
|
||||
}
|
||||
async function fence(tx: Parameters<Parameters<Db["transaction"]>[0]>[0], owner: NativeRunnerCheckpointFenceOwner | OfflineOwner,
|
||||
async function fence(tx: Parameters<Parameters<Db["transaction"]>[0]>[0], owner: NativeRunnerCheckpointFenceOwner | OfflineOwner | SettledControllerArchiveOwner,
|
||||
inactiveSource?: InactiveSource) {
|
||||
if (inactiveSource && "kind" in owner) fail("prestage_requires_runtime_owner");
|
||||
if ("kind" in owner && owner.kind === "settled_controller_archive") {
|
||||
await (await import("./native-controller-archive-source.js")).assertSettledControllerArchiveSource(tx as unknown as Db, owner);
|
||||
return;
|
||||
}
|
||||
const checkpointOwner = "kind" in owner ? null : await lockNativeRunnerCheckpointOwner(tx, owner);
|
||||
const [task] = await tx.select({ executionRunId: issues.executionRunId, checkoutRunId: issues.checkoutRunId }).from(issues)
|
||||
.where(and(eq(issues.companyId, owner.companyId), eq(issues.id, owner.issueId))).for("share");
|
||||
@@ -183,7 +188,7 @@ export async function assertNativeHarnessBackupOwner(db: Db, owner: NativeRunner
|
||||
}
|
||||
|
||||
export async function persistNativeHarnessBackup(input: {
|
||||
db: Db; storage: StorageService; owner: NativeRunnerCheckpointFenceOwner | OfflineOwner;
|
||||
db: Db; storage: StorageService; owner: NativeRunnerCheckpointFenceOwner | OfflineOwner | SettledControllerArchiveOwner;
|
||||
scopeDigest: string; root: string; signal?: AbortSignal; maxArchiveBytes?: number;
|
||||
inactiveSource?: { normalizedSessionId: string; runnerInstanceId: string; providerLeaseId: string };
|
||||
providerArtifacts?: NativeProviderArtifactSource;
|
||||
|
||||
Reference in new issue
Block a user