mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
fix(native): validate active controller durability during eligibility (#20601)
## Thinking Path > - Paperclip controls agents while their work runs outside the web server. > - A web replacement must retain each controller's durable recovery state. > - Modern controllers commit that state to the database before sending work. > - Their local controller file is only an external-persistence identity marker. > - Disk eligibility still required recovery metadata that can arrive after a run finishes. > - This pull request verifies the complete active database authority instead. > - It also bounds concurrent source reads so a large history does not require serial network waits. ## Linked Issues or Issue Description Refs: #20311, #20584, #20599. **What happened?** A live native run can have a complete database controller checkpoint and no terminal session snapshot yet. The disk-preservation check treats its local identity marker as a local-only obligation. The check also resolves selected sources one at a time. **Expected behavior** Accept a complete, bound active recovery authority. Refuse missing, partial, stale or foreign evidence. Keep all execution and enrollment guards. **Steps to reproduce** Use the new active-controller regression. Start a real Runner controller with the database checkpoint writer. Keep the business run active and omit its terminal snapshot metadata. The previous check returns a local obligation even after the full checkpoint commits. ## What Changed - Require the exact external-persistence marker and a complete decoded encrypted checkpoint for active runs. Match run, agent, task, session, Runner and workspace identities. For warm reuse, also permit an older marker only when its full older checkpoint matches that marker and its exact earlier successful run has a committed accepted result in the same company, agent, task, session, Runner and workspace. Keep the current live writer mandatory. Read the marker again after validation. - Accept that proof only with its fenced live writer and task binding, or the existing exact saved recovery snapshot. The saved-snapshot path still requires the full active journal. - Keep newer remote workspace authority strict. A malformed reference cannot fall back to an older prepared reference. Conflicting plugin task authority remains blocked. - Resolve selected sources at most eight at a time for read-only eligibility. Join every started reader before reporting an error. Keep source order, pagination, predicates and all write paths unchanged. Offline migration and prestaging still resolve sources one at a time. - Add real PostgreSQL regressions for active durability, partial and foreign state, expired ownership, previous prepared-reference state, post-result state and reader failure joining. ## Verification - Canonical Node24 preflight passed in74.2s. It built the current Runner inputs and published the cache. No shared Cargo target or hosted database was used. - Genuine active regression RED: the previous check returned true when false was required. The real checkpoint writer had committed the complete authority and replaced the local file with the Runner marker. Treatment passed, including the refusal controls. - Genuine source-reader RED: sixteen retained pointers used one reader at a time. Treatment used at most eight and joined all eight when the first reader failed. With a controlled20ms reader delay and real PostgreSQL selection, the small fixture changed from497.5ms to85.3ms. This is a bounded-concurrency proof, not a production speedup measurement. - All54 cases in the affected controller migration integration file passed in76.07s, with no skips. These include existing failed-controller, archived-controller, inline database and previous-session preservation cases. - Server TypeScript passed. The normal merge of current master kept both changed files byte-for-byte identical to the54-case proof. Final post-merge focused checks passed two of two in14.83s (52 other cases intentionally unselected); server TypeScript and module boundaries passed. Comment-only correction743 removed all new code comments required by CONTRIBUTING. A later warm-reuse regression reproduced the remaining false blocker through the real Runner rotateRunIdentity API. The bounded treatment uses one exact older-run lookup and full old/current checkpoint validation; it adds no Runner format, release or lineage service. Final normal-merge head `c62d6eb1755cf99b6def9563ba69854dba17de42` includes the new0459 baseline. All three focused PostgreSQL cases passed in21.37s (52 existing cases intentionally unselected); fresh server TypeScript and module boundaries passed. The prior54-case regression evidence is retained; no full matrix was repeated. - `git diff --check` passed. Final-head CI passed:56 checks succeeded and4 were skipped; the separate Paperclip Review advisory was still queued. All required test, build, typecheck and ci/verify gates passed on c62. [Final required workflow](https://github.com/paperclipai/paperclip/actions/runs/42868279612). Fresh [Greptile review](https://github.com/paperclipai/paperclip/pull/20601#issuecomment-6774537057) is5/5 on exact head `c62d6eb1755cf99b6def9563ba69854dba17de42`, with no open review threads. ```sh node scripts/agent-preflight.mjs --skip-rust-debug node --import ./cli/node_modules/tsx/dist/loader.mjs packages/db/src/test-postgres-command.ts pnpm exec vitest run server/src/services/native-runtime/native-controller-migration.integration.test.ts --disableConsoleIntercept pnpm --filter @paperclipai/server exec tsc --noEmit ``` ## Risks - This reports preservation evidence. It does not grant attachment, execution, replacement or enrollment authority. The normal live-owner and admission gates still apply. - A partial old or current checkpoint, stale local controller state, identity mismatch or uncertain owner stays blocked. Warm-marker proof also rejects a foreign agent, newer source, later completion, rejected result, per-turn scope or expired current writer. No source, checkpoint, session, lease, archive or provider state is changed by the check. - The eight readers add bounded database concurrency during an explicit eligibility request. There is no new background scan, cache, row filter or timeout change. Live enumeration time still needs deployed measurement. - The plugin task-lease branch uses the producer's environment lease row ID, as native-session-executor and existing diskless adoption coverage require. A direct active-eligibility positive for that plugin branch is a coverage limit; existing mismatch/conflicting-authority checks remain. The advisory found no must-fix defect, and this does not change the current tested revision. - Saved formats and terminal migration behavior remain unchanged. Older prepared workspace references remain supported. A fully saved result can still pass before the run-status update, but its entire active checkpoint must decode. - The observed staging run finished naturally and became valid on the existing terminal path during investigation. That observation is not deployed verification of the new active path. ## Model Used OpenAI GPT-6 (Codex), with reasoning, tool use and code execution. This runtime does not expose a more specific model ID or context-window size. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
9009fbdedb
commit
85dec04677
2 files changed
+290
-16
No files matched your search
+203
-7
@@ -1,15 +1,17 @@
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { mkdir, open, readFile, rm, stat, writeFile } from "node:fs/promises";
|
||||
import { dirname, join } from "node:path";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { afterEach, expect, it, vi } from "vitest";
|
||||
import { agentTaskSessions, environmentLeases, environments, heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, nativeRunResults, nativeRunnerCheckpoints } from "@paperclipai/db";
|
||||
import { agents, agentTaskSessions, environmentLeases, environments, heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, nativeRunResults, nativeRunnerCheckpoints } from "@paperclipai/db";
|
||||
import { captureLegacyDurablePrpCheckpoint, 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 * as sessionResume from "./native-session-resume.js";
|
||||
import { nativeControllerMigrationDescriptor } from "./native-session-executor.js";
|
||||
import { hasLocalNativeControllerObligations, migrateNativeControllers, prestageNativeControllers } from "./native-controller-migration.js";
|
||||
import { insertMigratedNativeRunnerCheckpoint, loadNativeRunnerCheckpoint } from "./native-runner-checkpoint.js";
|
||||
import { insertMigratedNativeRunnerCheckpoint, loadNativeRunnerCheckpoint, nativeRunnerCheckpointWriter } from "./native-runner-checkpoint.js";
|
||||
import { currentNativeControllerIdentity } from "./native-restart-recovery.js";
|
||||
import { createNativeControllerRecovery, resolveNativeControllerRecovery } from "./native-controller-recovery.js";
|
||||
import { resolveRemoteProviderPackAuthority } from "./remote-provider-pack-authority.js";
|
||||
import { canonicalProviderPackJson } from "./remote-provider-pack-verification.js";
|
||||
@@ -23,7 +25,7 @@ import { legacyNativeControllerArchive } from "./native-controller-archive.js";
|
||||
import { restoreDurableNativeHarnessBackup } from "./native-harness-backup-storage.js";
|
||||
|
||||
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", "durable_inline_failed",
|
||||
"active_warm", "source_scan", "active_durable", "online_success", "online_owned", "online_same_source", "online_busy", "online_changed", "online_evidence", "import", "archive", "success", "batch", "durable_failed", "durable_inline_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",
|
||||
@@ -32,11 +34,11 @@ describeEmbeddedPostgres.each(["archive_online", "archive_online_owned", "archiv
|
||||
"controller_observed_handoff", "controller_observed_handoff_stopped", "controller_observed_handoff_broken", "controller_observed_handoff_fork"])("offline native controller migration (%s)", scenario => {
|
||||
const environment = nativeRestartFixture();
|
||||
afterEach(() => { vi.unstubAllEnvs(); vi.restoreAllMocks(); });
|
||||
async function fixture(commandCount = 1, warm = false) {
|
||||
async function fixture(commandCount = 1, warm = false, modernWorkspace = false) {
|
||||
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 binding = { companyId, agentId, issueId: f.issueId, runId: f.runId, executionWorkspaceId: modernWorkspace ? randomUUID() : 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" },
|
||||
@@ -50,7 +52,7 @@ describeEmbeddedPostgres.each(["archive_online", "archive_online_owned", "archiv
|
||||
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 };
|
||||
environmentLeaseId: binding.executionWorkspaceId, 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();
|
||||
@@ -91,6 +93,200 @@ describeEmbeddedPostgres.each(["archive_online", "archive_online_owned", "archiv
|
||||
taskKey: f.issueId, lastRunId: f.runId }).returning();
|
||||
return { ...f, companyId, agentId, paperclipHome, runtimeRoot, paths, file, source, profile, session, queuedId, identity, snapshot };
|
||||
}
|
||||
if (scenario === "active_warm") it("preserves an active warm rotation only with its exact older committed authority", async () => {
|
||||
const f = await fixture(0, true, true), controller = await currentNativeControllerIdentity();
|
||||
const binding = { companyId: f.companyId, issueId: f.issueId, runId: f.runId };
|
||||
const owner = { controllerBootId: controller.bootId, controllerGeneration: 1, leaseOwner: "warm-checkpoint-writer",
|
||||
leaseExpiresAt: new Date(Date.now() + 60_000), phase: "observed", resultId: null, failureCode: null, recoveryState: null } as const;
|
||||
await f.db.update(nativeRunFinalizations).set(owner).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
const core = new DurablePrpControlPlane({ stateDirectory: join(f.paths.root, "control-plane"), identity: f.identity,
|
||||
expectedRunnerVersion: "migration-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` });
|
||||
let writer = nativeRunnerCheckpointWriter(f.db, { ...binding, bootId: controller.bootId, leaseOwner: owner.leaseOwner,
|
||||
generation: 1, authorityRunId: f.runId });
|
||||
core.setCheckpointPersistence(value => writer(value));
|
||||
try {
|
||||
await core.flushCheckpoint();
|
||||
const marker = await readFile(f.file, "utf8"), [prior] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
const [result] = await f.db.insert(nativeRunResults).values({ ...binding, completionContractId: prior.completionContractId!,
|
||||
serverFingerprint: "warm-source-result", schemaStatus: "accepted", resultJson: {}, canonicalSha256: nativeSha256({}) }).returning();
|
||||
const finishedAt = new Date();
|
||||
await f.db.update(heartbeatRuns).set({ status: "succeeded", finishedAt }).where(eq(heartbeatRuns.id, f.runId));
|
||||
await f.db.update(nativeRunFinalizations).set({ phase: "committed", resultId: result.id, leaseOwner: null,
|
||||
leaseExpiresAt: null }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
const next = { ...f.identity, runId: randomUUID(), turnId: randomUUID(), itemId: randomUUID() };
|
||||
const execution = parseNativeExecutionInput({ ...f.profile.nativeExecutionInput,
|
||||
binding: { ...f.profile.nativeExecutionInput.binding, runId: next.runId } });
|
||||
const profile = { ...f.profile, nativeExecutionInput: execution, sessionCheckpoint: null };
|
||||
expect(nativeControllerMigrationDescriptor(execution, f.profile.nativeToolContractFingerprint).root).toBe(f.paths.root);
|
||||
await f.db.insert(heartbeatRuns).values({ id: next.runId, companyId: f.companyId, agentId: f.agentId, issueId: f.issueId,
|
||||
nativeIssueId: f.issueId, nativeSessionId: f.native.normalizedSessionId, runnerInstanceId: f.native.runnerInstanceId,
|
||||
runtimeMode: "native", status: "running", nativePhase: "observed", processPid: process.pid, processStartedAt: new Date(),
|
||||
createdAt: new Date(finishedAt.getTime() + 1), startedAt: new Date(finishedAt.getTime() + 2), runnerProfileJson: profile,
|
||||
completionContractId: prior.completionContractId });
|
||||
await f.db.insert(nativeRunFinalizations).values({ ...owner, ...binding, runId: next.runId });
|
||||
await f.db.update(issues).set({ executionRunId: next.runId, checkoutRunId: null }).where(eq(issues.id, f.issueId));
|
||||
await f.db.update(agentTaskSessions).set({ lastRunId: next.runId }).where(eq(agentTaskSessions.id, f.session.id));
|
||||
writer = nativeRunnerCheckpointWriter(f.db, { ...binding, runId: next.runId, authorityRunId: next.runId,
|
||||
bootId: controller.bootId, leaseOwner: owner.leaseOwner, generation: 1 });
|
||||
core.rotateRunIdentity(next);
|
||||
await core.flushCheckpoint();
|
||||
expect(await readFile(f.file, "utf8")).toBe(marker);
|
||||
expect((await loadNativeRunnerCheckpoint(f.db, { ...binding, runId: next.runId }))?.runId).toBe(next.runId);
|
||||
const oldRows = await f.db.select().from(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId));
|
||||
const currentRows = await f.db.select().from(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, next.runId));
|
||||
expect(await hasLocalNativeControllerObligations(f.db)).toBe(false);
|
||||
for (const value of [{ ...f.identity, runId: randomUUID() }, { ...f.identity, runnerInstanceId: randomUUID() },
|
||||
{ ...f.identity, normalizedSessionId: randomUUID() }, { ...f.identity, environmentLeaseId: randomUUID() },
|
||||
{ ...f.identity, turnId: randomUUID() }]) {
|
||||
await writeFile(f.file, JSON.stringify({ schema: "paperclip.runner.external-control-plane-state.v1", identity: value }));
|
||||
expect(await hasLocalNativeControllerObligations(f.db)).toBe(true);
|
||||
}
|
||||
await writeFile(f.file, marker);
|
||||
await f.db.delete(nativeRunnerCheckpoints).where(and(eq(nativeRunnerCheckpoints.runId, f.runId), eq(nativeRunnerCheckpoints.part, "cursor")));
|
||||
expect(await hasLocalNativeControllerObligations(f.db)).toBe(true);
|
||||
await f.db.delete(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId));
|
||||
await f.db.insert(nativeRunnerCheckpoints).values(oldRows);
|
||||
await f.db.delete(nativeRunnerCheckpoints).where(and(eq(nativeRunnerCheckpoints.runId, next.runId), eq(nativeRunnerCheckpoints.part, "cursor")));
|
||||
expect(await hasLocalNativeControllerObligations(f.db)).toBe(true);
|
||||
await f.db.delete(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, next.runId));
|
||||
await f.db.insert(nativeRunnerCheckpoints).values(currentRows);
|
||||
for (const patch of [{ status: "failed" }, { createdAt: new Date(Date.now() + 60_000) },
|
||||
{ finishedAt: new Date(Date.now() + 60_000) }, { runnerInstanceId: randomUUID() }]) {
|
||||
await f.db.update(heartbeatRuns).set(patch).where(eq(heartbeatRuns.id, f.runId));
|
||||
expect(await hasLocalNativeControllerObligations(f.db)).toBe(true);
|
||||
await f.db.update(heartbeatRuns).set({ status: "succeeded", createdAt: prior.createdAt, finishedAt,
|
||||
runnerInstanceId: prior.runnerInstanceId }).where(eq(heartbeatRuns.id, f.runId));
|
||||
}
|
||||
await f.db.update(nativeRunResults).set({ schemaStatus: "rejected" }).where(eq(nativeRunResults.id, result.id));
|
||||
expect(await hasLocalNativeControllerObligations(f.db)).toBe(true);
|
||||
await f.db.update(nativeRunResults).set({ schemaStatus: "accepted" }).where(eq(nativeRunResults.id, result.id));
|
||||
const [foreign] = await f.db.insert(agents).values({ companyId: f.companyId, name: "Foreign owner", role: "engineer", adapterType: "paperclip_runner" }).returning();
|
||||
await f.db.update(heartbeatRuns).set({ agentId: foreign.id }).where(eq(heartbeatRuns.id, f.runId));
|
||||
expect(await hasLocalNativeControllerObligations(f.db)).toBe(true);
|
||||
await f.db.update(heartbeatRuns).set({ agentId: f.agentId }).where(eq(heartbeatRuns.id, f.runId));
|
||||
await f.db.update(heartbeatRuns).set({ runnerProfileJson: { ...profile, nativeExecutionInput: {
|
||||
...execution, session: { ...execution.session, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } } } } })
|
||||
.where(eq(heartbeatRuns.id, next.runId));
|
||||
expect(await hasLocalNativeControllerObligations(f.db)).toBe(true);
|
||||
await f.db.update(heartbeatRuns).set({ runnerProfileJson: profile }).where(eq(heartbeatRuns.id, next.runId));
|
||||
await f.db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(0) }).where(eq(nativeRunFinalizations.runId, next.runId));
|
||||
expect(await hasLocalNativeControllerObligations(f.db)).toBe(true);
|
||||
await f.db.update(nativeRunFinalizations).set(owner).where(eq(nativeRunFinalizations.runId, next.runId));
|
||||
expect(await hasLocalNativeControllerObligations(f.db)).toBe(false);
|
||||
expect(await readFile(f.file, "utf8")).toBe(marker);
|
||||
expect(await f.db.select().from(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId))).toEqual(oldRows);
|
||||
} finally { await core.stop(); }
|
||||
});
|
||||
if (scenario === "source_scan") it("bounds retained source readers at eight and joins every started reader before a failure", async () => {
|
||||
const f = await fixture(), binding = { companyId: f.companyId, issueId: f.issueId, runId: f.runId };
|
||||
const checkpoint = captureLegacyDurablePrpCheckpoint(f.source, f.identity,
|
||||
{ version: "migration-fixture", digest: `sha256:${"a".repeat(64)}` });
|
||||
await f.db.transaction(tx => insertMigratedNativeRunnerCheckpoint(tx, binding, checkpoint));
|
||||
await f.db.update(heartbeatRuns).set({ runnerProfileJson: { ...f.profile,
|
||||
nativeControllerRecovery: createNativeControllerRecovery(f.snapshot, { configDigest: f.paths.configDigest,
|
||||
processConfigDigest: null }, { runId: f.runId, runnerInstanceId: f.native.runnerInstanceId,
|
||||
normalizedSessionId: f.native.normalizedSessionId }) } }).where(eq(heartbeatRuns.id, f.runId));
|
||||
await f.db.insert(agentTaskSessions).values(Array.from({ length: 15 }, () => ({ companyId: f.companyId,
|
||||
agentId: f.agentId, adapterType: "paperclip_runner", taskKey: randomUUID(), lastRunId: f.runId })));
|
||||
const original = sessionResume.findNativeSessionResumeRun;
|
||||
let started = 0, finished = 0, inFlight = 0, peak = 0, fail = false;
|
||||
const reason = new Error("retained source reader failed");
|
||||
const spy = vi.spyOn(sessionResume, "findNativeSessionResumeRun").mockImplementation(async (...args) => {
|
||||
const first = ++started === 1; peak = Math.max(peak, ++inFlight);
|
||||
try {
|
||||
await new Promise(resolve => setTimeout(resolve, 20));
|
||||
if (fail && first) throw reason;
|
||||
return await original(...args);
|
||||
} finally { inFlight--; finished++; }
|
||||
});
|
||||
try {
|
||||
const before = performance.now();
|
||||
expect(await hasLocalNativeControllerObligations(f.db)).toBe(false);
|
||||
console.info(JSON.stringify({ state: "retained_source_readers", elapsedMs: performance.now() - before, peak, started, finished }));
|
||||
expect(started).toBe(16); expect(finished).toBe(16); expect(peak).toBeGreaterThan(1); expect(peak).toBeLessThanOrEqual(8);
|
||||
started = finished = peak = 0; fail = true;
|
||||
await expect(hasLocalNativeControllerObligations(f.db)).rejects.toBe(reason);
|
||||
expect(started).toBe(8); expect(finished).toBe(8); expect(inFlight).toBe(0);
|
||||
expect(await readFile(f.file, "utf8")).toBe(f.source);
|
||||
} finally { spy.mockRestore(); }
|
||||
});
|
||||
if (scenario === "active_durable") it("accepts a complete active database authority without terminal metadata and refuses stale or incomplete handoff proof", async () => {
|
||||
const f = await fixture(1, false, true), controller = await currentNativeControllerIdentity();
|
||||
const binding = { companyId: f.companyId, issueId: f.issueId, runId: f.runId };
|
||||
const { nativeWorkspaceSync, ...baseProfile } = f.profile;
|
||||
const remoteWorkspace = { schema: "paperclip.remote-workspace-run.v1", workspaceId: nativeWorkspaceSync.workspaceId,
|
||||
leaseId: nativeWorkspaceSync.leaseId, providerLeaseId: nativeWorkspaceSync.providerLeaseId,
|
||||
remoteCwd: nativeWorkspaceSync.remoteCwd, epoch: 1, state: "prepared", snapshotId: null };
|
||||
const profile = { ...baseProfile, sessionCheckpoint: null, remoteWorkspace };
|
||||
await f.db.update(heartbeatRuns).set({ status: "running", finishedAt: null, nativePhase: "observed",
|
||||
processPid: process.pid, processStartedAt: new Date(), runnerProfileJson: profile }).where(eq(heartbeatRuns.id, f.runId));
|
||||
const owner = { controllerBootId: controller.bootId, controllerGeneration: 1, leaseOwner: "live-checkpoint-writer",
|
||||
leaseExpiresAt: new Date(Date.now() + 60_000), phase: "observed", resultId: null, failureCode: null, recoveryState: null } as const;
|
||||
await f.db.update(nativeRunFinalizations).set(owner).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
await f.db.update(issues).set({ executionRunId: f.runId, checkoutRunId: null }).where(eq(issues.id, f.issueId));
|
||||
const core = new DurablePrpControlPlane({ stateDirectory: join(f.paths.root, "control-plane"), identity: f.identity,
|
||||
expectedRunnerVersion: "migration-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` });
|
||||
core.setCheckpointPersistence(nativeRunnerCheckpointWriter(f.db, { ...binding, bootId: controller.bootId,
|
||||
leaseOwner: owner.leaseOwner, generation: 1, authorityRunId: f.runId }));
|
||||
try {
|
||||
await core.flushCheckpoint();
|
||||
const marker = await readFile(f.file, "utf8");
|
||||
const rows = await f.db.select().from(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId));
|
||||
expect(rows.length).toBeGreaterThan(2);
|
||||
expect(await hasLocalNativeControllerObligations(f.db)).toBe(false);
|
||||
for (const value of [f.source, JSON.stringify({ schema: "unknown", identity: f.identity }),
|
||||
JSON.stringify({ schema: "paperclip.runner.external-control-plane-state.v1", identity: { ...f.identity, runId: randomUUID() } })]) {
|
||||
await writeFile(f.file, value); expect(await hasLocalNativeControllerObligations(f.db)).toBe(true);
|
||||
}
|
||||
await writeFile(f.file, marker);
|
||||
for (const part of ["cursor", "command:1"]) {
|
||||
await f.db.delete(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.part, part));
|
||||
expect(await hasLocalNativeControllerObligations(f.db)).toBe(true);
|
||||
const savedProfile = { ...profile, sessionCheckpoint: f.snapshot,
|
||||
nativeControllerRecovery: createNativeControllerRecovery(f.snapshot, { configDigest: f.paths.configDigest,
|
||||
processConfigDigest: null }, { runId: f.runId, runnerInstanceId: f.native.runnerInstanceId,
|
||||
normalizedSessionId: f.native.normalizedSessionId }) };
|
||||
await f.db.update(heartbeatRuns).set({ runnerProfileJson: savedProfile }).where(eq(heartbeatRuns.id, f.runId));
|
||||
expect(await hasLocalNativeControllerObligations(f.db)).toBe(true);
|
||||
await f.db.delete(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId));
|
||||
await f.db.insert(nativeRunnerCheckpoints).values(rows);
|
||||
await f.db.update(nativeRunFinalizations).set({ phase: "workspace_finalizing", leaseOwner: null })
|
||||
.where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
expect(await hasLocalNativeControllerObligations(f.db)).toBe(false);
|
||||
await f.db.update(nativeRunFinalizations).set(owner).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
await f.db.update(heartbeatRuns).set({ runnerProfileJson: profile }).where(eq(heartbeatRuns.id, f.runId));
|
||||
}
|
||||
const cipher = createLocalEncryptedCipher("write"), header = JSON.parse((await loadNativeRunnerCheckpoint(f.db, binding))!.records.get("header")!);
|
||||
for (const field of ["runId", "runnerInstanceId", "normalizedSessionId", "environmentLeaseId"]) {
|
||||
const contents = JSON.stringify({ ...header, state: { ...header.state, identity: { ...header.state.identity, [field]: randomUUID() } } });
|
||||
await f.db.update(nativeRunnerCheckpoints).set({ material: cipher.encrypt(JSON.stringify({ ...binding, part: "header", contents })) })
|
||||
.where(eq(nativeRunnerCheckpoints.part, "header"));
|
||||
expect(await hasLocalNativeControllerObligations(f.db)).toBe(true);
|
||||
}
|
||||
await f.db.delete(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId));
|
||||
await f.db.insert(nativeRunnerCheckpoints).values(rows);
|
||||
for (const patch of [{ leaseExpiresAt: new Date(0) }, { controllerGeneration: 0 }, { controllerBootId: null },
|
||||
{ leaseOwner: null }, { failureCode: "uncertain" }, { phase: "retryable_failure" }]) {
|
||||
await f.db.update(nativeRunFinalizations).set(patch).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
expect(await hasLocalNativeControllerObligations(f.db)).toBe(true);
|
||||
await f.db.update(nativeRunFinalizations).set(owner).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
}
|
||||
await f.db.update(issues).set({ executionRunId: f.queuedId }).where(eq(issues.id, f.issueId));
|
||||
expect(await hasLocalNativeControllerObligations(f.db)).toBe(true);
|
||||
await f.db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId));
|
||||
for (const value of [null, {}, { ...remoteWorkspace, workspaceId: randomUUID() }]) {
|
||||
await f.db.update(heartbeatRuns).set({ runnerProfileJson: { ...profile, remoteWorkspace: value,
|
||||
nativeWorkspaceSync } }).where(eq(heartbeatRuns.id, f.runId));
|
||||
expect(await hasLocalNativeControllerObligations(f.db)).toBe(true);
|
||||
}
|
||||
await f.db.update(heartbeatRuns).set({ runnerProfileJson: { ...baseProfile, sessionCheckpoint: null, nativeWorkspaceSync } })
|
||||
.where(eq(heartbeatRuns.id, f.runId));
|
||||
expect(await hasLocalNativeControllerObligations(f.db)).toBe(false);
|
||||
await f.db.update(heartbeatRuns).set({ runnerProfileJson: profile }).where(eq(heartbeatRuns.id, f.runId));
|
||||
expect(await readFile(f.file, "utf8")).toBe(marker);
|
||||
expect(await f.db.select().from(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId))).toEqual(rows);
|
||||
} finally { await core.stop(); }
|
||||
});
|
||||
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));
|
||||
|
||||
@@ -3,9 +3,10 @@ import { lstat, open, readdir } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import { and, eq, inArray, or, sql } from "drizzle-orm";
|
||||
import { heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db";
|
||||
import { remoteWorkspaceRunReferenceSchema } from "@paperclipai/shared";
|
||||
import { captureLegacyDurablePrpCheckpoint, inspectDurablePrpCheckpointHeader, parseNativeExecutionInput } from "../../vendor/paperclip-runner/index.js";
|
||||
import { offlineMigrationExecutionWhere } from "../offline-migration-execution.js";
|
||||
import { findNativeSessionResumeRun } from "./native-session-resume.js";
|
||||
import { findNativeSessionResumeRun, isNativeSessionId } 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";
|
||||
@@ -76,7 +77,7 @@ async function hasConfirmedOlderController(db: Db, selected: Selected,
|
||||
} catch { return false; }
|
||||
}
|
||||
|
||||
async function* retainedSources(db: Db, signal?: AbortSignal) {
|
||||
async function* retainedSources(db: Db, signal?: AbortSignal, concurrency = 1) {
|
||||
let after: string | undefined;
|
||||
for (;;) {
|
||||
signal?.throwIfAborted();
|
||||
@@ -86,12 +87,18 @@ async function* retainedSources(db: Db, signal?: AbortSignal) {
|
||||
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) {
|
||||
for (let offset = 0; offset < rows.length; offset += concurrency) {
|
||||
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 };
|
||||
const batch = await Promise.allSettled(rows.slice(offset, offset + concurrency).map(async selected => ({
|
||||
selected, source: await findNativeSessionResumeRun(db, { companyId: selected.companyId, agentId: selected.agentId,
|
||||
issueId: selected.issueId, normalizedSessionId: selected.sessionId,
|
||||
beforeCreatedAt: new Date(selected.createdAt), includeProgressBarrier: true }),
|
||||
})));
|
||||
for (const result of batch) {
|
||||
signal?.throwIfAborted();
|
||||
if (result.status === "rejected") throw result.reason;
|
||||
if (result.value.source) yield { selected: result.value.selected, source: result.value.source };
|
||||
}
|
||||
}
|
||||
if (rows.length < 100) return;
|
||||
after = rows.at(-1)!.id;
|
||||
@@ -175,18 +182,89 @@ async function preservedDatabaseOnlyFailedCheckpoint(selected: Selected,
|
||||
return true;
|
||||
}
|
||||
|
||||
async function preservedWarmMarker(db: Db, selected: Selected,
|
||||
source: NonNullable<Awaited<ReturnType<typeof findNativeSessionResumeRun>>>, markerIdentity: Record<string, unknown>) {
|
||||
if (!isNativeSessionId(markerIdentity.runId) || markerIdentity.runId === source.id) return false;
|
||||
const [prior] = await db.execute<{ id: string }>(sql`
|
||||
select r.id from heartbeat_runs r join heartbeat_runs current on current.id=${source.id}::uuid and current.company_id=r.company_id
|
||||
join native_run_finalizations f on f.run_id=r.id and f.company_id=r.company_id and f.issue_id=r.native_issue_id
|
||||
join native_run_results result on result.id=f.result_id and result.run_id=r.id and result.company_id=r.company_id
|
||||
and result.issue_id=r.native_issue_id and result.completion_contract_id=r.completion_contract_id and result.schema_status='accepted'
|
||||
where r.id=${markerIdentity.runId}::uuid and r.company_id=${selected.companyId}::uuid and r.agent_id=${selected.agentId}::uuid
|
||||
and r.native_issue_id=${selected.issueId}::uuid and r.native_session_id=${selected.sessionId}::uuid
|
||||
and r.runner_instance_id=${source.runnerInstanceId} and r.runtime_mode='native' and r.status='succeeded'
|
||||
and r.created_at<current.created_at and r.finished_at is not null and r.finished_at<=coalesce(current.started_at,current.created_at)
|
||||
and f.phase='committed' and f.lease_owner is null and f.recovery_state is null and f.failure_code is null`);
|
||||
if (!prior) return false;
|
||||
const checkpoint = await loadNativeRunnerCheckpoint(db, { companyId: selected.companyId, issueId: selected.issueId, runId: prior.id }).catch(() => null);
|
||||
const header = checkpoint && inspectDurablePrpCheckpointHeader(checkpoint.records.get("header") ?? "");
|
||||
return Boolean(header && nativeSha256(header) === nativeSha256(markerIdentity));
|
||||
}
|
||||
|
||||
async function preservedActiveCheckpoint(db: Db, selected: Selected,
|
||||
source: NonNullable<Awaited<ReturnType<typeof findNativeSessionResumeRun>>>, paths: Awaited<ReturnType<typeof localPaths>>) {
|
||||
if (source.status !== "running" || source.runtimeMode !== "native") return false;
|
||||
let marker: Record<string, unknown>, bytes: string | undefined;
|
||||
try {
|
||||
bytes = await readPrivateSource(paths.controlFile, 4096);
|
||||
marker = record(JSON.parse(bytes ?? "null"));
|
||||
} catch { return false; }
|
||||
if (marker.schema !== "paperclip.runner.external-control-plane-state.v1"
|
||||
|| Object.keys(marker).sort().join(",") !== "identity,schema") return false;
|
||||
const binding = { companyId: selected.companyId, issueId: selected.issueId, runId: source.id };
|
||||
const checkpoint = await loadNativeRunnerCheckpoint(db, binding).catch(() => null);
|
||||
const header = checkpoint && inspectDurablePrpCheckpointHeader(checkpoint.records.get("header") ?? "");
|
||||
if (!header || header.runId !== source.id || header.runnerInstanceId !== source.runnerInstanceId
|
||||
|| header.normalizedSessionId !== selected.sessionId) return false;
|
||||
const exact = nativeSha256(header) === nativeSha256(marker.identity), markerIdentity = record(marker.identity);
|
||||
if (!exact && (paths.execution.session.lifecyclePolicy.mode !== "warm"
|
||||
|| ["runnerInstanceId", "normalizedSessionId", "environmentLeaseId"].some(field => markerIdentity[field] !== header[field as keyof typeof header])
|
||||
|| !await preservedWarmMarker(db, selected, source, markerIdentity))) return false;
|
||||
const [current] = await db.execute<{ workspace: unknown; remoteWorkspace: unknown; hasRemoteWorkspace: boolean; pluginLeaseId: string | null; liveOwner: boolean }>(sql`
|
||||
select 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",
|
||||
(r.native_phase='observed' and r.process_pid is not null and r.process_started_at is not null
|
||||
and i.execution_run_id=r.id and i.assignee_agent_id=r.agent_id
|
||||
and f.controller_boot_id is not null and f.controller_boot_id<>'' and f.controller_generation>0
|
||||
and f.lease_owner is not null and f.lease_owner<>'' and f.lease_expires_at>clock_timestamp()
|
||||
and f.phase='observed' and f.result_id is null and f.failure_code is null) is true as "liveOwner"
|
||||
from heartbeat_runs r left join native_run_finalizations f on f.run_id=r.id and f.company_id=r.company_id and f.issue_id=r.native_issue_id
|
||||
left join issues i on i.id=r.native_issue_id and i.company_id=r.company_id
|
||||
left join lateral (select l.id from environment_leases l where l.company_id=r.company_id and l.heartbeat_run_id=r.id
|
||||
and l.issue_id=r.native_issue_id and l.status='active' and l.released_at is null and l.cleanup_status is null
|
||||
and (l.expires_at is null or l.expires_at>clock_timestamp()) and l.metadata->>'agentId'=r.agent_id::text
|
||||
and l.metadata->>'runnerTaskLease'='true' and l.metadata->>'runnerTaskDispatched'='true'
|
||||
and coalesce(l.metadata->>'runnerTaskExecutionStopped','false')!='true'
|
||||
and coalesce(l.metadata->>'runnerTaskRetiring','false')!='true' order by l.created_at desc limit 1) p on true
|
||||
where r.id=${source.id}::uuid and r.company_id=${selected.companyId}::uuid and r.agent_id=${selected.agentId}::uuid
|
||||
and r.native_issue_id=${selected.issueId}::uuid and r.native_session_id=${selected.sessionId}::uuid
|
||||
and r.runner_instance_id=${source.runnerInstanceId} and r.runtime_mode='native' and r.status='running'`);
|
||||
if (!current) return false;
|
||||
const remote = remoteWorkspaceRunReferenceSchema.safeParse(current.remoteWorkspace);
|
||||
const workspaceId = current.hasRemoteWorkspace
|
||||
? remote.success && !current.pluginLeaseId ? remote.data.workspaceId : null
|
||||
: current.pluginLeaseId ?? readNativeWorkspaceSyncReference(current.workspace)?.workspaceId;
|
||||
return workspaceId === header.environmentLeaseId
|
||||
&& (current.pluginLeaseId !== null || paths.execution.binding.executionWorkspaceId === workspaceId)
|
||||
&& (current.liveOwner || exact && hasMatchingControllerRecovery(selected, source, checkpoint))
|
||||
&& bytes === await readPrivateSource(paths.controlFile, 4096);
|
||||
}
|
||||
|
||||
export async function hasLocalNativeControllerObligations(db: Db, storage?: StorageService | null): 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)) {
|
||||
for await (const { selected, source } of retainedSources(db, undefined, 8)) {
|
||||
const binding = { companyId: selected.companyId, issueId: selected.issueId, runId: source.id };
|
||||
if (readNativeControllerRecovery(source.runnerProfileJson.nativeControllerRecovery, source.runnerProfileJson.sessionCheckpoint)
|
||||
if (source.status !== "running"
|
||||
&& readNativeControllerRecovery(source.runnerProfileJson.nativeControllerRecovery, source.runnerProfileJson.sessionCheckpoint)
|
||||
&& await hasNativeRunnerCheckpoint(db, binding)) continue;
|
||||
if (source.status === "failed" && source.runnerProfileJson.sessionCheckpoint == null
|
||||
&& preservedFailedCheckpoint(source, selected.sessionId, await loadNativeRunnerCheckpoint(db, binding))) continue;
|
||||
const paths = await localPaths(source);
|
||||
if (await preservedActiveCheckpoint(db, selected, source, paths)) continue;
|
||||
const present = await lstat(paths.controlFile).catch((error: NodeJS.ErrnoException) => {
|
||||
if (error.code === "ENOENT") return undefined; throw error;
|
||||
});
|
||||
|
||||
Reference in new issue
Block a user