mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-08 22:13:59 +01:00
fix(runner): retry proven ingress bootstrap failures (#18981)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Native execution must retain its admitted model and execution authority during recovery. > - A sandbox provider can reject Runner endpoint acquisition before the Runner gets a credential. > - The old error path loses the structured HTTP status and leaves that startup failure ambiguous. > - A zero counter in an old checkpoint cannot prove that no credential was issued. > - This pull request adds a versioned producer proof and carries closed provider status through the existing RPC. > - A proved transient startup failure uses the existing retry budget. A terminal proved-unused run can settle its accounting. ## Linked Issues or Issue Description Refs #18018 and #18934. Those changes restore existing checkpoints and ready retained environments. This change covers endpoint acquisition before Runner dispatch. A search of related public PRs found no duplicate implementation. **What happened?** A provider HTTP 502 during Runner endpoint acquisition becomes an untyped worker error. Native recovery treats it as ambiguous and stops without retrying. The run's native usage receipt also stays pending after startup ends without model work. **Expected behavior** Retry a structured transient endpoint failure only when the complete current controller checkpoint proves that this exact run never issued a Runner credential. Preserve authentication, configuration, Stop, ownership and provider-progress gates. Settle zero usage only under the existing terminal accounting owner. **Steps to reproduce** 1. Use the controlled endpoint in the Daytona ingress integration test to return HTTP 502. 2. Register the actual native Runner driver and persist its bootstrap journal through the encrypted checkpoint writer. 3. Observe the old terminal ambiguous disposition instead of bounded startup recovery. **Paperclip version or commit** Reproduced on the pre-change master source. The original regression used the production native driver, a real HTTP fault, a real durable control plane and local Postgres. The new contract additionally tests the pinned Daytona SDK and real worker RPC. ## What Changed - The controller owns a v2 persisted state with retained origin and credential history. Current readers still accept v1. Older writers reject v2. Restore and checkpoint decoding do not add missing history. - The proof validates the complete bootstrap journal, all six identity fields and every delivery, credential and progress control. The server rechecks its fingerprint under the current controller claim before retry or accounting changes. - Daytona and the SDK carry a closed acquisition phase and HTTP status through the existing ingress RPC. Both public error wrappers omit response bodies and causes, including Pino cause serialization. - Only structured HTTP 429 or 5xx ingress failures can use the existing three-attempt startup budget. HTTP 401/403, configuration and unknown errors do not gain retry authority. - A terminal proved-unused run wakes existing accounting and records zero provider work. It does not create a semantic result, prove physical termination or release an unknown process/workspace hold. - The normal automatic second attempt rechecks its exact claim, frozen execution and matching complete local/encrypted journal before admitting absent remote Runner state. It keeps the original pending bootstrap command IDs. A proved unused Codex home can rematerialize sealed skills. Full raw local state is checked before streaming-event or policy projection. - Normal saved-state readers and the existing E2E journal evidence matcher accept the closed v1/v2 set. The separate legacy no-launch witness is unchanged. Technical docs describe the contracts. ## Verification Current head: `cd38f70409cdb8664e14bfbf4efa208f468a0419` (final delta adds only fixture artifact preparation). Production files are unchanged from the independently reviewed `93c8a33520` source. Full repository typecheck and build passed at `96417acb58`; the affected server typecheck also passes after the final two-line fixture delta. Focused protocol, raw-proof, real retry, SDK and RPC acceptance pass on this source. Earlier broad suite counts are listed with their limits below: - Durable Runner protocol: 187 tests. - Native executor, driver, startup/accounting and redaction: 1,211 tests, including real Postgres and encrypted checkpoint persistence. - Normal saved-state restore/projection compatibility: 21 focused cases. - SDK: 209 tests. - Standalone Daytona: 544 tests; 8 opt-in live tests skipped. - Explicit standalone `pnpm -C packages/plugins/sandbox-providers/daytona exec vitest run --config vitest.sdk-ingress.config.ts`: 6 cases with real pinned SDK 0.203.0 and worker RPC ingress acceptance. These include lookup 502/401/403/400, private-preview 502 after successful lookup/refresh, missing credentials and private-response exclusion. - E2E journal evidence: 4 cases across both supported formats, with wrong ownership, missing stimulus and unknown-schema controls. - The exact historical controller source rejects a new v2 checkpoint at decode and restore. A v1 control is accepted. This verifies the downgrade fence with the actual older parser. - Startup controls cover issued and pruned credentials, old/truncated journals, provider progress, exhausted retry accounting, and checkpoint/Stop/controller-claim races. Race cases retain the current execution and pending usage. - New exact admission controls: 19 cases through encrypted persistence and cold restore. They cover current/stale/expired claims, Stop, pause, model/trust/account changes, prior progress, issued/pruned credentials, adopted/old/truncated state, local/canonical fingerprint mismatch, hidden delivery, unsupported work and company scope. Two additional red/green cases prove that raw local stream progress and missing event policy are rejected before lossy checkpoint projection. No case mutates run, claim, result or accounting state. - Actual native executor, scheduler and Rust Runner retry: 2 cases. Codex uses its fake app server. OpenCode uses the real proxy and fake OpenCode HTTP server with the frozen DeepSeek model and allow permission configuration. Each path launches no Runner on attempt 1, starts one Runner and provider on attempt 2, reuses both original command IDs, commits one accepted result, validates the persisted local workspace, records nonzero synthetic native usage and settles accounting once. No zero-work accounting flag survives. The controlled SDK and protocol tests use local source, isolated PostgreSQL, a private HTTPS preview and fake provider servers on macOS arm64 with Node 24.22.2. They create no hosted sandbox and make no model call. The root workspace intentionally omits standalone provider dependencies; the explicit SDK target fails if its pinned dependency is absent and has no conditional skip. They do not qualify hosted rollout or repair an old unmarked checkpoint. The first full local typecheck found missing checkout SDK links from an install with scripts disabled; the repository link helper repaired that setup. Current-head Linux CI typecheck and build pass. Independent final review checked 12 source scopes and six raw-proof controls with no remaining finding. Fresh Greptile review is 5/5 with no unresolved thread. The full Runner CI suite found two older control-peer test doubles missing the new proof method. Both now return no proof, with all 27 permission, ordering and cancellation assertions retained and passing. The new Cargo-dependent native retry suite uses the existing Rust-cached CI lane. All 39 lane coverage controls pass and prove complete, duplicate-free suite coverage. The first full test run stopped at an admitted chat Retry polling failure. CI also found a stale retained-allocation fixture, now repaired on master by #18984, and a canceled runner job. A real SDK suite initially relied on a dependency absent from the root workspace; the explicit standalone target now owns that acceptance. The separate Product E2E support run had 12 failures in four untouched files. All 12 reproduce on the original clean base with independent dependencies, compiled base source and verified workspace resolution. These concern workflow expectations, cleanup timeout arguments, a profile snapshot and operating receipts. They are not passing acceptance. The later full local repository test command stopped at a 15-second timeout in the unchanged GitHub pre-allocation retry `controller` case. That exact case passed in isolation with unchanged assertions. The broad Runner run initially omitted `test:typescript:prep` and lacked debug fake-provider binaries. Required preparation then passed 370 + 44 package contract cases (11 skipped); the prepared full package run reported 4,932 passed, 9 failed and 13 skipped. All failed families then passed with unchanged assertions in isolation: 8 ACPX controller tests and 2 credential controls. The paired original-base ACPX control also passed. The full-run failures remain recorded; they are not passing full-suite evidence. The first Rust-cached native CI run found a missing generated OpenCode proxy in a fresh checkout. The fixture now runs the existing verified provider-entrypoint builder. Removing the generated proxy reproduces the exact ENOENT before the fix; both real retry cases pass after the fix from the same absent-artifact state. All current-head CI gates pass, including the aggregate verification check. Linux native CI passes all 10 suites: 177 tests pass and 14 opt-in cases skip. Both real provider retry cases pass. The full Runner package passes 297 suites and 4,950 tests, with 1 suite and 10 tests skipped. The CLI workspace groups pass 4,762 and 286 tests. The first CI attempt had Runner shutdowns in the Runner package, Canary and three general server jobs. One unchanged CLI shard also hit child-server startup and dense ZIP deadlines. Both CLI files passed unchanged in isolation: 18 tests. One bounded rerun of unsuccessful jobs and their dependents passed. The original failures remain recorded. ## Risks - Older controllers reject newly written v2 state on downgrade. Current controllers retain v1 restore support, but old or unmarked state cannot gain this new proof. - An older plugin that omits structured ingress evidence remains held. The server and provider must use the new producer contract for this recovery path. The Docker `cloud` target builds the SDK, server/Runner and standalone Daytona producer from one source tree. The default `production` target does not build the standalone producer. Hosted rollout must confirm the new image and active image-owned plugin worker; a registry version alone does not prove its source. An npm-installed producer needs an explicit matching artifact upgrade. - This prevents future proved startup failures. It does not release historical accounting based on an old checkpoint, change retry budgets, replace a model or weaken sandbox isolation. - Zero model usage is separate from process termination, saved files and resource release. Those owners keep their existing gates. ## Model Used OpenAI Codex with tool use and code execution. This runtime does not expose the exact backend model ID or context window to the agent. Read-only independent agents reviewed checkpoint compatibility and SDK/RPC error boundaries. ## 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 (affected suites; full local run limits are recorded above) - [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
6a45ea479e
commit
3a2249dd57
34 files changed
+1243
-57
No files matched your search
@@ -143,6 +143,7 @@ export const PAPERCLIP_PUBLIC_SCHEMA_IDS = new Set([
|
||||
"paperclip.runner.conformance.fixture.v1",
|
||||
"paperclip.runner.conformance.output.v1",
|
||||
"paperclip.runner.durable.control-plane-state.v1",
|
||||
"paperclip.runner.durable.control-plane-state.v2",
|
||||
"paperclip.runner.durable.diagnostics.v1",
|
||||
"paperclip.runner.durable.state.v1",
|
||||
"paperclip.runner.durable.trace.v1",
|
||||
|
||||
@@ -0,0 +1,124 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { readFile } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { expect, it } from "vitest";
|
||||
import { agents, completionContracts, heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations,
|
||||
nativeRunnerCheckpoints, nativeRunResults } from "@paperclipai/db";
|
||||
import { DurablePrpControlPlane, parseNativeExecutionInput } from "../../vendor/paperclip-runner/index.js";
|
||||
import { describeEmbeddedPostgres, nativeRestartFixture } from "../../__tests__/helpers/native-restart.js";
|
||||
import { admitUnclaimedNativeBootstrapRetry } from "./native-bootstrap-retry.js";
|
||||
import { buildNativeCompletionContract } from "./completion-contracts.js";
|
||||
import { loadNativeRunnerCheckpoint, nativeRunnerCheckpointWriter } from "./native-runner-checkpoint.js";
|
||||
import { nativeRuntimeContextFixture } from "./runtime-context.test-fixture.js";
|
||||
|
||||
describeEmbeddedPostgres("unclaimed native bootstrap retry admission", () => {
|
||||
const environment = nativeRestartFixture();
|
||||
it.each([
|
||||
"current proof", "stale claim", "expired claim", "Stop", "paused agent", "changed model", "changed trust",
|
||||
"changed account", "provider progress", "issued and pruned credential", "adopted authority", "legacy state",
|
||||
"truncated checkpoint", "local fingerprint mismatch", "local streamed progress", "local missing event policy",
|
||||
"hidden delivery", "unknown command", "foreign company",
|
||||
])("admits only a current complete proof: %s", async scenario => {
|
||||
const { db, runId, issueId, native } = await environment.seedRun(`ADMISSION-${scenario}`);
|
||||
const { companyId, agentId, runtimeRoot } = environment.current;
|
||||
const [issue] = await db.select().from(issues).where(eq(issues.id, issueId));
|
||||
const [contract] = await db.select().from(completionContracts).where(eq(completionContracts.issueId, issueId));
|
||||
const execution = parseNativeExecutionInput({
|
||||
schema: "paperclip.native-execution-input.v5", executionMode: "default", planningContext: null,
|
||||
runtimeContext: nativeRuntimeContextFixture(),
|
||||
binding: { companyId, agentId, issueId, runId, executionWorkspaceId: runId },
|
||||
provider: { kind: "codex", model: "fixture-model", approvalPolicy: "never" },
|
||||
task: { identifier: "fixture", title: issue!.title, description: null, prompt: "Controlled admission fixture.", workMode: "standard" },
|
||||
workspace: { cwd: runtimeRoot, repoUrl: null, repoRef: null, branchName: null },
|
||||
session: { normalizedSessionId: native.normalizedSessionId, driverKind: "codex_app_server", protocolVersion: 1,
|
||||
lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } },
|
||||
completionContract: { id: contract!.id, sha256: contract!.canonicalSha256, schemaVersion: "paperclip.completion-contract.v1",
|
||||
contract: buildNativeCompletionContract(issue!, { revision: contract!.revision }) }, interactionResponses: [], credentialBindings: [],
|
||||
});
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: sql`coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb)
|
||||
|| ${JSON.stringify({ nativeExecutionInput: execution })}::jsonb` }).where(eq(heartbeatRuns.id, runId));
|
||||
const leaseOwner = `fixture:${randomUUID()}`, bootId = randomUUID();
|
||||
await db.update(nativeRunFinalizations).set({ leaseOwner, controllerBootId: bootId, controllerGeneration: 7,
|
||||
leaseExpiresAt: new Date(Date.now() + 60_000), attempt: 1 }).where(eq(nativeRunFinalizations.runId, runId));
|
||||
const identity = { runnerInstanceId: native.runnerInstanceId, environmentLeaseId: runId, runId,
|
||||
normalizedSessionId: native.normalizedSessionId, turnId: `turn-${runId}`, itemId: `item-${runId}` };
|
||||
const root = join(runtimeRoot, runId, "control-plane");
|
||||
const core = new DurablePrpControlPlane({ stateDirectory: root, identity, expectedRunnerVersion: "0.3.0",
|
||||
expectedRunnerDigest: `sha256:${"a".repeat(64)}` });
|
||||
const template = { provider: { provider: "codex", command: "fixture" }, authorizedTools: {} };
|
||||
core.persistRunAttachTemplate(template);
|
||||
core.queueCommand("run.prepare", template);
|
||||
core.queueCommand("session.open", { reuse: "same_session" });
|
||||
try {
|
||||
if (scenario === "issued and pruned credential") { core.issueBootstrapTicket(); core.store.state.tickets = {}; }
|
||||
if (scenario === "adopted authority") core.store.state.bootstrapHistory!.adopted = true;
|
||||
if (scenario === "legacy state") {
|
||||
core.store.state.schema = "paperclip.runner.durable.control-plane-state.v1";
|
||||
delete core.store.state.bootstrapHistory;
|
||||
}
|
||||
if (scenario === "hidden delivery") core.store.state.commandDeliveryCounts = { removedCommand: 1 };
|
||||
if (scenario === "unknown command") core.queueCommand("turn.start", { prompt: "Prior work must stay held." });
|
||||
core.setCheckpointPersistence(nativeRunnerCheckpointWriter(db, {
|
||||
companyId, issueId, runId, authorityRunId: runId, bootId, leaseOwner, generation: 7,
|
||||
}));
|
||||
await core.flushCheckpoint();
|
||||
const projection = join(runtimeRoot, runId, "projection");
|
||||
const checkpoint = await loadNativeRunnerCheckpoint(db, execution.binding);
|
||||
expect(checkpoint).not.toBeNull();
|
||||
checkpoint!.restoreSessionDirectory(projection);
|
||||
let localStateContents = await readFile(join(projection, "control-plane", "control-plane-state.json"), "utf8");
|
||||
const original = JSON.parse(localStateContents);
|
||||
if (scenario === "local streamed progress" || scenario === "local missing event policy") {
|
||||
expect(await admitUnclaimedNativeBootstrapRetry({ db, execution, leaseOwner, localStateContents })).toBe(true);
|
||||
if (scenario === "local missing event policy") delete original.checkpointEventPolicy;
|
||||
else {
|
||||
const event = { schema: "paperclip.prp.event.v1", schemaVersion: 1, sourceKind: "runner",
|
||||
sourceInstanceId: identity.runnerInstanceId, sourceEventId: "fixture-local-progress", sourceSeq: 1, runId,
|
||||
normalizedSessionId: identity.normalizedSessionId, turnId: identity.turnId, itemId: identity.itemId,
|
||||
eventType: "item.delta", priority: 2, emittedAt: "2026-01-01T00:00:00.000Z", payload: { delta: "Prior local progress" } };
|
||||
original.committedEvents = [{ sourceSeq: 1, sourceEventId: event.sourceEventId, eventType: event.eventType,
|
||||
priority: event.priority, envelope: { kind: "event", payload: event }, deliveryCount: 1, logicalEffectCount: 1 }];
|
||||
}
|
||||
localStateContents = JSON.stringify(original);
|
||||
}
|
||||
if (scenario === "local fingerprint mismatch") {
|
||||
original.commands[1].payload = { reuse: "same_session", changed: true };
|
||||
localStateContents = JSON.stringify(original);
|
||||
}
|
||||
if (scenario === "truncated checkpoint") await db.delete(nativeRunnerCheckpoints).where(and(
|
||||
eq(nativeRunnerCheckpoints.runId, runId), eq(nativeRunnerCheckpoints.part, "command:2")));
|
||||
if (scenario === "provider progress") await db.insert(heartbeatRunEvents).values({ companyId, runId, agentId, seq: 900,
|
||||
eventType: "turn.started", stream: "system", level: "info", message: "Prior provider progress", payload: {} });
|
||||
if (scenario === "stale claim") await db.update(nativeRunFinalizations).set({ leaseOwner: "successor",
|
||||
controllerGeneration: 8 }).where(eq(nativeRunFinalizations.runId, runId));
|
||||
if (scenario === "expired claim") await db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(0) })
|
||||
.where(eq(nativeRunFinalizations.runId, runId));
|
||||
if (scenario === "Stop") await db.update(heartbeatRuns).set({ resultJson: {
|
||||
nativeCancellation: { schema: "paperclip.native-cancellation.v1", companyId, issueId, runId,
|
||||
scope: "run", dispatchState: "pending", intentAuditId: randomUUID() },
|
||||
} }).where(eq(heartbeatRuns.id, runId));
|
||||
if (scenario === "paused agent") await db.update(agents).set({ status: "paused" }).where(eq(agents.id, agentId));
|
||||
const proposed = structuredClone(execution);
|
||||
if (scenario === "changed model") proposed.provider.model = "different-model";
|
||||
if (scenario === "changed trust" && proposed.provider.kind === "codex") proposed.provider.approvalPolicy = "on-request";
|
||||
if (scenario === "changed account") proposed.credentialBindings = [{ bindingId: "different-account", service: "openai",
|
||||
destination: "environment", expiresAt: null, displayName: null }];
|
||||
if (scenario === "foreign company") proposed.binding.companyId = randomUUID();
|
||||
const [beforeRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
|
||||
const [beforeClaim] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId));
|
||||
expect(await admitUnclaimedNativeBootstrapRetry({ db, execution: proposed, leaseOwner, localStateContents }))
|
||||
.toBe(scenario === "current proof");
|
||||
const [afterRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
|
||||
const [afterClaim] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId));
|
||||
expect(afterRun).toEqual(beforeRun);
|
||||
expect(afterClaim).toEqual(beforeClaim);
|
||||
expect(afterRun.usageJson?.accountingProviderWorkStarted).not.toBe(false);
|
||||
expect(afterRun.costAccountedAt).toBeNull();
|
||||
expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, runId))).toEqual([]);
|
||||
} finally {
|
||||
await core.stop();
|
||||
if (scenario === "paused agent") await db.update(agents).set({ status: "active" }).where(eq(agents.id, agentId));
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,51 @@
|
||||
import { and, eq, isNull, sql } from "drizzle-orm";
|
||||
import { heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db";
|
||||
import { inspectUnclaimedBootstrapState, parseNativeExecutionInput,
|
||||
type NativeExecutionInput } from "../../vendor/paperclip-runner/index.js";
|
||||
import { nativeSha256 } from "./canonical.js";
|
||||
import { loadNativeUnclaimedBootstrapCheckpoint } from "./native-runner-checkpoint.js";
|
||||
import { isNativeRunnerOwnershipHeld } from "./native-runner-ownership.js";
|
||||
import { isUnusedNativeSessionBootstrap, nativeSessionProviderEvidence } from "./native-session-resume.js";
|
||||
import { readNativeSessionResumptionAgent } from "./native-session-resumption-admission.js";
|
||||
|
||||
export async function admitUnclaimedNativeBootstrapRetry(input: {
|
||||
db: Db;
|
||||
execution: NativeExecutionInput;
|
||||
leaseOwner: string;
|
||||
localStateContents: string;
|
||||
preparedExecutionAdmission?: { admissionSha256: string; preparedSha256: string };
|
||||
}): Promise<boolean> {
|
||||
const { binding } = input.execution;
|
||||
return input.db.transaction(async tx => {
|
||||
await tx.execute(sql`select set_config('statement_timeout', '15000', true), set_config('lock_timeout', '1000', true)`);
|
||||
const [issue] = await tx.select().from(issues).where(and(
|
||||
eq(issues.id, binding.issueId), eq(issues.companyId, binding.companyId))).for("update");
|
||||
const [coordinator] = await tx.select().from(nativeRunFinalizations).where(and(
|
||||
eq(nativeRunFinalizations.runId, binding.runId), eq(nativeRunFinalizations.companyId, binding.companyId),
|
||||
eq(nativeRunFinalizations.issueId, binding.issueId), eq(nativeRunFinalizations.leaseOwner, input.leaseOwner),
|
||||
eq(nativeRunFinalizations.phase, "observed"), isNull(nativeRunFinalizations.resultId))).for("update");
|
||||
const [run] = await tx.select().from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.id, binding.runId), eq(heartbeatRuns.companyId, binding.companyId),
|
||||
eq(heartbeatRuns.agentId, binding.agentId), eq(heartbeatRuns.nativeIssueId, binding.issueId))).for("update");
|
||||
if (!issue || !coordinator || !run || coordinator.attempt < 1 || coordinator.attempt >= 3
|
||||
|| !coordinator.leaseExpiresAt || coordinator.leaseExpiresAt.getTime() <= Date.now()
|
||||
|| run.runtimeMode !== "native" || run.status !== "running" || run.finishedAt !== null
|
||||
|| !run.runnerInstanceId || !run.nativeSessionId || run.nativeSessionId !== input.execution.session.normalizedSessionId
|
||||
|| isNativeRunnerOwnershipHeld(run) || run.resultJson?.nativeCancellation != null
|
||||
|| run.resultJson?.startupCancellation != null
|
||||
|| !isUnusedNativeSessionBootstrap(run, (await nativeSessionProviderEvidence(tx, [run.id])).has(run.id))) return false;
|
||||
if (!await readNativeSessionResumptionAgent(tx as unknown as Db, { issue, run, coordinator })) return false;
|
||||
let admitted: NativeExecutionInput;
|
||||
try { admitted = parseNativeExecutionInput(run.runnerProfileJson?.nativeExecutionInput); }
|
||||
catch { return false; }
|
||||
const executionSha256 = nativeSha256(input.execution);
|
||||
if (nativeSha256(admitted) !== (input.preparedExecutionAdmission?.admissionSha256 ?? executionSha256)
|
||||
|| input.preparedExecutionAdmission && executionSha256 !== input.preparedExecutionAdmission.preparedSha256) return false;
|
||||
const identity = { runnerInstanceId: run.runnerInstanceId, environmentLeaseId: binding.executionWorkspaceId,
|
||||
runId: binding.runId, normalizedSessionId: run.nativeSessionId, turnId: `turn-${binding.runId}`, itemId: `item-${binding.runId}` };
|
||||
const canonical = await loadNativeUnclaimedBootstrapCheckpoint(tx, binding, identity);
|
||||
if (!canonical) return false;
|
||||
return inspectUnclaimedBootstrapState(input.localStateContents, identity,
|
||||
{ version: canonical.runnerVersion, digest: canonical.runnerDigest })?.fingerprint === canonical.fingerprint;
|
||||
});
|
||||
}
|
||||
+281
@@ -0,0 +1,281 @@
|
||||
import { execFileSync, type ChildProcess } from "node:child_process";
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { readFileSync } from "node:fs";
|
||||
import { copyFile, mkdir, readFile, stat, writeFile } from "node:fs/promises";
|
||||
import https, { createServer as createHttpsServer } from "node:https";
|
||||
import { createServer as createTcpServer } from "node:net";
|
||||
import { join, resolve } from "node:path";
|
||||
import { getCACertificates, setDefaultCACertificates } from "node:tls";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import WebSocket, { WebSocketServer, type RawData } from "ws";
|
||||
import { beforeAll, expect, it, vi } from "vitest";
|
||||
import { completionContracts, executionWorkspaces, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults } from "@paperclipai/db";
|
||||
import type { RunnerIngressEndpoint } from "@paperclipai/adapter-utils/runner-connectivity";
|
||||
import { environmentRunnerIngressErrorData, JsonRpcCallError, PluginEnvironmentRunnerIngressError } from "@paperclipai/plugin-sdk";
|
||||
import { canonicalNativeRuntimeContextDigest, createNativeSessionBackend, createRunnerdCodexTransport, parseNativeExecutionInput,
|
||||
type RunnerProcessHandle } from "../../vendor/paperclip-runner/index.js";
|
||||
import { nativeRestartFixture, describeEmbeddedPostgres, stopOwnedProcessGroup } from "../../__tests__/helpers/native-restart.js";
|
||||
import { canonicalNativeJson, nativeSha256 } from "./canonical.js";
|
||||
import { executePaperclipNativeSession, nativeSessionWorkspaceScope, resolveNativeBilling } from "./native-session-executor.js";
|
||||
import { createNativeRunnerExecutionDriver } from "./native-runner-execution-driver.js";
|
||||
import { claimNativeSessionResumptions, reconcileNativeFinalizations } from "./native-finalization-reconciler.js";
|
||||
import { currentNativeControllerIdentity } from "./native-restart-recovery.js";
|
||||
import { loadNativeRunnerCheckpoint } from "./native-runner-checkpoint.js";
|
||||
import { buildNativeCompletionContract } from "./completion-contracts.js";
|
||||
import { releaseMaterializedNativeRuntimeSkills, spawnRunner } from "../../vendor/paperclip-runner/testing.js";
|
||||
import { materializeAsset } from "./runtime-context.js";
|
||||
import { nativeRuntimeContextFixture } from "./runtime-context.test-fixture.js";
|
||||
import { createNativeUsageDelivery } from "./native-usage-delivery.js";
|
||||
import { accountRunCost } from "../run-cost-accounting.js";
|
||||
|
||||
describeEmbeddedPostgres("native ingress bootstrap retry through the real Runner", () => {
|
||||
const environment = nativeRestartFixture();
|
||||
const runnerWorkspace = resolve(import.meta.dirname, "../../../../packages/paperclip-runner/runner");
|
||||
const runnerBinary = join(runnerWorkspace, "target/release/paperclip-runnerd");
|
||||
const fakeProvider = join(runnerWorkspace, "target/release/fake-codex-app-server");
|
||||
|
||||
beforeAll(() => {
|
||||
execFileSync(process.execPath, [resolve(runnerWorkspace, "../scripts/build-verified-provider-entrypoints.mjs")],
|
||||
{ cwd: resolve(runnerWorkspace, ".."), stdio: "pipe", timeout: 30_000 });
|
||||
execFileSync("cargo", ["build", "--release", "--locked", "-p", "paperclip-runner-core",
|
||||
"--bin", "paperclip-runnerd", "--bin", "fake-codex-app-server"],
|
||||
{ cwd: runnerWorkspace, stdio: "ignore", timeout: 600_000 });
|
||||
}, 660_000);
|
||||
|
||||
it.each(["codex", "opencode"] as const)("reacquires %s ingress and commits exactly one result on the automatic second attempt", async provider => {
|
||||
const fixture = await environment.seedRun(`INGRESS-BOOTSTRAP-RETRY-${provider}`);
|
||||
const { db, runId, issueId, native } = fixture;
|
||||
const { companyId, agentId, runtimeRoot } = environment.current;
|
||||
const [issue] = await db.select().from(issues).where(eq(issues.id, issueId));
|
||||
const [contract] = await db.select().from(completionContracts).where(eq(completionContracts.issueId, issueId));
|
||||
const sandbox = join(runtimeRoot, `ingress-sandbox-${provider}`);
|
||||
const stateBase = join(runtimeRoot, `ingress-controller-${provider}`);
|
||||
const codexHome = join(runtimeRoot, `ingress-codex-home-${provider}`);
|
||||
const filesystem = join(runtimeRoot, `ingress-filesystem-${provider}`);
|
||||
await mkdir(sandbox, { recursive: true });
|
||||
await mkdir(codexHome, { recursive: true });
|
||||
await mkdir(join(filesystem, "codex-home"), { recursive: true });
|
||||
await db.insert(executionWorkspaces).values({ id: runId, companyId, sourceIssueId: issueId,
|
||||
mode: "isolated", strategyType: "directory", name: "Owned retry fixture", cwd: sandbox });
|
||||
await db.update(issues).set({ executionWorkspaceId: runId }).where(eq(issues.id, issueId));
|
||||
const runtimeContext = nativeRuntimeContextFixture();
|
||||
const skillContent = "---\nname: retry-fixture\ndescription: Controlled sealed retry skill.\n---\nKeep the admitted task and configuration.\n";
|
||||
runtimeContext.instructions.bundle = await materializeAsset([{ path: "AGENTS.md", mode: 0o444,
|
||||
content: Buffer.from("Complete only the controlled fixture task.\n") }]);
|
||||
if (provider === "codex") runtimeContext.skills = [{ key: "company/retry-fixture", runtimeName: "retry-fixture",
|
||||
versionId: "fixture-v1", bundle: await materializeAsset([{ path: "SKILL.md", mode: 0o444, content: Buffer.from(skillContent) }]) }];
|
||||
runtimeContext.aggregateDigest = canonicalNativeRuntimeContextDigest(runtimeContext);
|
||||
const execution = parseNativeExecutionInput({
|
||||
schema: "paperclip.native-execution-input.v5", executionMode: "default", planningContext: null, runtimeContext,
|
||||
binding: { companyId, agentId, issueId, runId, executionWorkspaceId: runId },
|
||||
provider: provider === "codex" ? { kind: "codex", model: "fixture-model", approvalPolicy: "never" }
|
||||
: { kind: "opencode", model: "openrouter/deepseek/deepseek-v4.1-flash-1017", permissionMode: "allow" },
|
||||
task: { identifier: "fixture", title: issue!.title, description: null,
|
||||
prompt: "Complete the controlled fixture task.", workMode: "standard" },
|
||||
workspace: { cwd: sandbox, repoUrl: null, repoRef: null, branchName: null },
|
||||
session: { normalizedSessionId: native.normalizedSessionId, driverKind: provider === "codex" ? "codex_app_server" : "opencode_server", protocolVersion: 1,
|
||||
lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } },
|
||||
completionContract: { id: contract!.id, sha256: contract!.canonicalSha256, schemaVersion: "paperclip.completion-contract.v1",
|
||||
contract: buildNativeCompletionContract(issue!, { revision: contract!.revision }) }, interactionResponses: [], credentialBindings: [],
|
||||
});
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: sql`coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb)
|
||||
|| ${JSON.stringify({ nativeExecutionInput: execution })}::jsonb` }).where(eq(heartbeatRuns.id, runId));
|
||||
const root = join(stateBase, nativeSha256({ schema: "paperclip.native-session-scope.v2", companyId, agentId,
|
||||
workspace: nativeSessionWorkspaceScope(execution), provider: { driverKind: execution.session.driverKind,
|
||||
identity: { kind: provider } }, normalizedSessionId: native.normalizedSessionId }));
|
||||
const digest = (file: string) => `sha256:${createHash("sha256").update(readFileSync(file)).digest("hex")}`;
|
||||
const opencode = join(runtimeRoot, "fake-opencode-ingress");
|
||||
const providerCalls = join(sandbox, "provider-calls.log");
|
||||
if (provider === "opencode") {
|
||||
const script = resolve(runnerWorkspace, "../test/fixtures/fake-opencode-server.mjs");
|
||||
execFileSync("cc", ["-x", "c", "-o", opencode, "-"], { input:
|
||||
`#include <unistd.h>\n#include <stdlib.h>\n#include <stdio.h>\nint main(int argc, char **argv) { FILE *log = fopen(${JSON.stringify(providerCalls)}, "a"); if (!log) return 126; fputs("process-start\\n", log); fclose(log); char **args = calloc(argc + 2, sizeof(char *)); args[0] = ${JSON.stringify(process.execPath)}; args[1] = ${JSON.stringify(script)}; for (int i = 1; i < argc; i++) args[i + 1] = argv[i]; execv(args[0], args); return 127; }` });
|
||||
}
|
||||
const proxy = resolve(runnerWorkspace, "../dist/cli/opencode-app-server-proxy.cjs");
|
||||
const identity = { runnerInstanceId: native.runnerInstanceId, environmentLeaseId: runId, runId,
|
||||
normalizedSessionId: native.normalizedSessionId, turnId: `turn-${runId}`, itemId: `item-${runId}` };
|
||||
const assertSealedSkills = async () => {
|
||||
if (provider !== "codex") return;
|
||||
const skill = join(root, "codex-home", "skills", "retry-fixture", "SKILL.md");
|
||||
expect(await readFile(skill, "utf8")).toBe(skillContent);
|
||||
expect((await stat(skill)).mode & 0o222).toBe(0);
|
||||
expect((await stat(join(root, "codex-home", "skills", "retry-fixture"))).mode & 0o222).toBe(0);
|
||||
};
|
||||
const resultFile = join(sandbox, "fixture-result.json");
|
||||
await writeFile(resultFile, canonicalNativeJson({ schema: "paperclip.run_result.v1", reportedWorkDisposition: "done",
|
||||
summary: "Controlled retry completed.", completionClaim: { contractRevision: execution.completionContract.contract.revision,
|
||||
objectiveSatisfied: true, criteria: execution.completionContract.contract.criteria.map(criterion => ({
|
||||
criterionId: criterion.id, status: "satisfied", evidenceRefs: [] })), remainingWork: [] },
|
||||
evidence: [], verification: [], artifacts: [], attentionRequests: [] }));
|
||||
const key = join(sandbox, "preview.key"), certificate = join(sandbox, "preview.crt");
|
||||
execFileSync("openssl", ["req", "-x509", "-newkey", "rsa:2048", "-nodes", "-keyout", key, "-out", certificate,
|
||||
"-days", "1", "-subj", "/CN=localhost", "-addext", "subjectAltName=IP:127.0.0.1,DNS:localhost"], { stdio: "ignore" });
|
||||
const previousCertificates = getCACertificates("default");
|
||||
const certificates = [...previousCertificates, await readFile(certificate, "utf8")];
|
||||
setDefaultCACertificates(certificates);
|
||||
const previousHttpsAgent = https.globalAgent;
|
||||
const fixtureHttpsAgent = new https.Agent({ ca: certificates });
|
||||
https.globalAgent = fixtureHttpsAgent;
|
||||
const portProbe = createTcpServer();
|
||||
await new Promise<void>(done => portProbe.listen(0, "127.0.0.1", done));
|
||||
const address = portProbe.address();
|
||||
if (!address || typeof address === "string") throw new Error("Missing fixture port");
|
||||
const runnerPort = address.port;
|
||||
await new Promise<void>(done => portProbe.close(() => done()));
|
||||
let healthy = false, acquisitions = 0, previewUrl = "";
|
||||
const preview = createHttpsServer({ key: await readFile(key), cert: await readFile(certificate) }, (request, response) => {
|
||||
if (request.url !== "/preview") { response.writeHead(404).end(); return; }
|
||||
acquisitions++;
|
||||
response.writeHead(healthy ? 200 : 502, { "content-type": "application/json" });
|
||||
response.end(JSON.stringify(healthy ? { url: previewUrl, token: "fixture-preview" } : { message: "fixture upstream unavailable" }));
|
||||
});
|
||||
const sockets = new Set<WebSocket>();
|
||||
const websocket = new WebSocketServer({ noServer: true });
|
||||
preview.on("upgrade", (request, socket, head) => {
|
||||
if (request.url !== `/api/runner/v1/connect/${runId}` || request.headers["x-fixture-preview-token"] !== "fixture-preview") {
|
||||
socket.end("HTTP/1.1 403 Forbidden\r\n\r\n"); return;
|
||||
}
|
||||
websocket.handleUpgrade(request, socket, head, front => {
|
||||
const back = new WebSocket(`ws://127.0.0.1:${runnerPort}${request.url}`, { perMessageDeflate: false });
|
||||
const pending: Array<{ data: RawData; binary: boolean }> = [];
|
||||
sockets.add(front); sockets.add(back);
|
||||
front.on("message", (data, binary) => {
|
||||
if (back.readyState === WebSocket.OPEN) back.send(data, { binary });
|
||||
else pending.push({ data, binary });
|
||||
});
|
||||
back.on("open", () => { for (const message of pending.splice(0)) back.send(message.data, { binary: message.binary }); });
|
||||
back.on("message", (data, binary) => { if (front.readyState === WebSocket.OPEN) front.send(data, { binary }); });
|
||||
front.on("close", () => { sockets.delete(front); back.terminate(); });
|
||||
back.on("close", () => { sockets.delete(back); front.terminate(); });
|
||||
front.on("error", error => { diagnostics.push(`fixture front: ${error.message}`); back.terminate(); });
|
||||
back.on("error", error => { diagnostics.push(`fixture back: ${error.message}`); front.terminate(); });
|
||||
});
|
||||
});
|
||||
preview.on("tlsClientError", error => { diagnostics.push(`fixture TLS: ${error.message}`); });
|
||||
await new Promise<void>(done => preview.listen(0, "127.0.0.1", done));
|
||||
const previewAddress = preview.address();
|
||||
if (!previewAddress || typeof previewAddress === "string") throw new Error("Missing preview address");
|
||||
previewUrl = `wss://127.0.0.1:${previewAddress.port}/api/runner/v1/connect/${runId}`;
|
||||
const bundles: ReturnType<typeof createRunnerdCodexTransport>[] = [];
|
||||
const handles: RunnerProcessHandle[] = [];
|
||||
const diagnostics: string[] = [];
|
||||
let usageDelivery: Awaited<ReturnType<typeof createNativeUsageDelivery>> | undefined;
|
||||
vi.stubEnv("PAPERCLIP_RUNNER_STATE_DIR", stateBase);
|
||||
const endpoint: () => Promise<RunnerIngressEndpoint> = async () => {
|
||||
const response = await fetch(`https://127.0.0.1:${previewAddress.port}/preview`);
|
||||
if (!response.ok) {
|
||||
const error = new PluginEnvironmentRunnerIngressError({ phase: "private_preview", httpStatus: response.status });
|
||||
throw new JsonRpcCallError({ code: -32000, message: error.message, data: environmentRunnerIngressErrorData(error) });
|
||||
}
|
||||
return { kind: "authenticated_websocket" as const, websocketUrl: previewUrl,
|
||||
secretHeaders: [{ name: "X-Fixture-Preview-Token", value: "fixture-preview" }], generation: "fixture",
|
||||
refresh: endpoint, close: async () => {} };
|
||||
};
|
||||
const target = { kind: "remote", transport: "sandbox", remoteCwd: sandbox, leaseId: randomUUID(),
|
||||
effectiveCapabilities: { runnerWebSocketIngress: true }, getRunnerIngressEndpoint: endpoint } as never;
|
||||
const backend = () => createNativeSessionBackend(execution, {
|
||||
runnerInstanceId: native.runnerInstanceId, environment: { CODEX_HOME: codexHome, PATH: process.env.PATH },
|
||||
codexTransportFactory: () => {
|
||||
let started!: () => void;
|
||||
const processStarted = new Promise<void>(done => { started = done; });
|
||||
const bundle = createRunnerdCodexTransport({
|
||||
provider, runnerBinary, codexCommand: fakeProvider, codexArgs: ["--state-file", join(sandbox, "provider.json"),
|
||||
"--terminal-result-file", resultFile, "--call-log", providerCalls, "--record-process-start"],
|
||||
stateDirectory: root, runnerStateDirectory: join(sandbox, "runner"), runnerFilesystemRoot: filesystem,
|
||||
runtimeContext, runnerRuntimeContext: runtimeContext, closeGraceMs: 1_000,
|
||||
onDiagnostic: message => { diagnostics.push(message); },
|
||||
onAccountingEvent: async event => { await usageDelivery?.capture(event); },
|
||||
environment: { CODEX_HOME: codexHome, PATH: process.env.PATH, OPENROUTER_API_KEY: "synthetic-fixture-key" }, sourceCodexHome: codexHome,
|
||||
...(provider === "opencode" ? { opencodeCommand: opencode, opencodeCommandSha256: digest(opencode),
|
||||
opencodeProxyPath: proxy, opencodeProxySha256: digest(proxy), providerNodeCommand: process.execPath,
|
||||
providerNodeCommandSha256: digest(process.execPath), opencodePermissionMode: "allow" as const } : {}),
|
||||
prpIdentity: identity, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null },
|
||||
controlPlaneRegistration: async authority => {
|
||||
const controller = await currentNativeControllerIdentity();
|
||||
return createNativeRunnerExecutionDriver({ target, binding: execution.binding,
|
||||
runnerIngressAuthorized: true,
|
||||
prepareRemote: async () => { await copyFile(join(root, "runtime-context.json"), join(filesystem, "runtime-context.json")); },
|
||||
processStarted,
|
||||
onLog: async (_stream, message) => { diagnostics.push(message); },
|
||||
controllerPersistence: { db, bootId: controller.bootId } }).controlPlaneRegistration!(authority);
|
||||
},
|
||||
runnerProcessLauncher: spec => {
|
||||
const handle = spawnRunner({ ...spec.nativeRunner!, runnerBinaryPath: spec.command, environment: spec.environment,
|
||||
connection: { mode: "listen", listenAddress: "0.0.0.0", listenPort: runnerPort,
|
||||
listenPath: `/api/runner/v1/connect/${runId}` } });
|
||||
(handle.child as ChildProcess).once("spawn", started);
|
||||
handles.push(handle); return handle;
|
||||
},
|
||||
});
|
||||
bundles.push(bundle); return bundle.transport;
|
||||
},
|
||||
});
|
||||
const execute = (leaseOwner?: string) => executePaperclipNativeSession({ db, execution,
|
||||
runnerInstanceId: native.runnerInstanceId, useRunnerd: true, runnerExecutionTarget: target,
|
||||
backend: backend(), leaseOwner, turnTimeoutMs: 15_000 });
|
||||
try {
|
||||
await expect(execute()).rejects.toThrow("HTTP 502");
|
||||
expect(acquisitions).toBe(1);
|
||||
expect(handles).toHaveLength(0);
|
||||
await assertSealedSkills();
|
||||
const saved = await loadNativeRunnerCheckpoint(db, execution.binding);
|
||||
expect(saved?.records.size).toBe(4);
|
||||
const savedBootstrapIds = [...saved!.records].filter(([part]) => part.startsWith("command:"))
|
||||
.map(([, value]) => JSON.parse(value).commandId);
|
||||
expect(savedBootstrapIds).toHaveLength(2);
|
||||
const [beforeRetry] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
|
||||
expect(beforeRetry.usageJson?.accountingProviderWorkStarted).not.toBe(false);
|
||||
expect(beforeRetry.costAccountedAt).toBeNull();
|
||||
const [failed] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId));
|
||||
expect(failed).toMatchObject({ phase: "retryable_failure", attempt: 1, failureDetail: { recoveryMode: "bootstrap_retry" } });
|
||||
const [claim] = await claimNativeSessionResumptions({ db, runnerInstanceId: "fixture-resumption", runIds: [runId],
|
||||
now: new Date(failed!.nextAttemptAt!.getTime() + 1), deferIneligible: false });
|
||||
expect(claim).toBeDefined();
|
||||
usageDelivery = await createNativeUsageDelivery(db, { companyId, runId, runnerInstanceId: native.runnerInstanceId,
|
||||
provider: execution.provider, billing: resolveNativeBilling(execution.provider) });
|
||||
healthy = true;
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
const retry = execute(claim!.leaseOwner);
|
||||
try {
|
||||
await expect(Promise.race([retry, new Promise<never>((_done, reject) => {
|
||||
timer = setTimeout(() => reject(new Error(`Controlled retry deadline exceeded: ${diagnostics.slice(-12).join("\n")}`)), 20_000);
|
||||
})])).resolves.toMatchObject({ exitCode: 0, timedOut: false });
|
||||
} finally { if (timer) clearTimeout(timer); }
|
||||
expect(acquisitions).toBe(2);
|
||||
expect(handles).toHaveLength(1);
|
||||
await assertSealedSkills();
|
||||
const completed = await loadNativeRunnerCheckpoint(db, execution.binding);
|
||||
const commands = [...completed!.records].filter(([part]) => part.startsWith("command:")).map(([, value]) => JSON.parse(value));
|
||||
expect(commands.filter(command => command.type === "run.prepare")).toHaveLength(1);
|
||||
expect(commands.filter(command => command.type === "session.open")).toHaveLength(1);
|
||||
expect(commands.filter(command => ["run.prepare", "session.open"].includes(command.type))
|
||||
.map(command => command.commandId)).toEqual(savedBootstrapIds);
|
||||
expect((await readFile(providerCalls, "utf8")).split("\n").filter(line => line === "process-start")).toHaveLength(1);
|
||||
const results = await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, runId));
|
||||
expect(results).toHaveLength(1);
|
||||
expect(results[0]).toMatchObject({ schemaStatus: "accepted" });
|
||||
await reconcileNativeFinalizations(db, [runId]);
|
||||
const [settled] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId));
|
||||
expect(settled).toMatchObject({ phase: "committed", attempt: 2 });
|
||||
const [completedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
|
||||
expect(completedRun.usageJson?.accountingProviderWorkStarted).not.toBe(false);
|
||||
expect(completedRun.usageJson?.accountingReceiptReady).toBe(true);
|
||||
expect(Number(completedRun.usageJson?.inputTokens)).toBeGreaterThan(0);
|
||||
expect(await accountRunCost(db, runId)).toBe(true);
|
||||
expect(await accountRunCost(db, runId)).toBe(false);
|
||||
const [accountedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
|
||||
expect(accountedRun.costAccountedAt).toBeInstanceOf(Date);
|
||||
} finally {
|
||||
for (const bundle of bundles) await bundle.transport.close().catch(() => undefined);
|
||||
for (const handle of handles) await stopOwnedProcessGroup(handle.child.pid ?? null, sandbox);
|
||||
if (provider === "codex") await releaseMaterializedNativeRuntimeSkills(join(root, "codex-home", "skills"));
|
||||
for (const socket of sockets) socket.terminate();
|
||||
websocket.close(); preview.closeAllConnections();
|
||||
await new Promise<void>(done => preview.close(() => done()));
|
||||
https.globalAgent = previousHttpsAgent;
|
||||
fixtureHttpsAgent.destroy();
|
||||
setDefaultCACertificates(previousCertificates);
|
||||
vi.unstubAllEnvs();
|
||||
}
|
||||
}, 60_000);
|
||||
});
|
||||
@@ -1,7 +1,7 @@
|
||||
import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js";
|
||||
import { and, eq, exists, gt, isNull, lte, inArray, notExists, or, sql } from "drizzle-orm";
|
||||
import { checkpointDb, heartbeatRuns, issues, nativeRunFinalizations, nativeRunnerCheckpoints, type Db } from "@paperclipai/db";
|
||||
import { inspectDurablePrpCheckpointHeader, decodeDurablePrpCheckpoint, type DurablePrpCheckpoint, type DurablePrpControlPlane } from "../../vendor/paperclip-runner/index.js";
|
||||
import { inspectDurablePrpCheckpointHeader, inspectUnclaimedBootstrapCheckpoint, decodeDurablePrpCheckpoint, type DurablePrpCheckpoint, type DurablePrpControlPlane, type DurableRecoveryIdentity } from "../../vendor/paperclip-runner/index.js";
|
||||
import { createLocalEncryptedCipher } from "../../secrets/local-encrypted-provider.js";
|
||||
import { legacyControllerBootId } from "../legacy-controller-lease.js";
|
||||
|
||||
@@ -226,6 +226,22 @@ export async function loadNativeRunnerCheckpoint(db: Pick<Db, "select">, binding
|
||||
return checkpoint;
|
||||
}
|
||||
|
||||
export async function loadNativeUnclaimedBootstrapCheckpoint(
|
||||
db: Pick<Db, "select">, binding: Binding, identity: DurableRecoveryIdentity,
|
||||
) {
|
||||
try {
|
||||
if (binding.runId !== identity.runId) return null;
|
||||
const rows = await db.select().from(nativeRunnerCheckpoints).where(and(
|
||||
eq(nativeRunnerCheckpoints.companyId, binding.companyId), eq(nativeRunnerCheckpoints.issueId, binding.issueId),
|
||||
eq(nativeRunnerCheckpoints.runId, binding.runId))).limit(5);
|
||||
if (rows.length !== 4 || rows.reduce((bytes, row) => bytes + Buffer.byteLength(JSON.stringify(row.material)), 0)
|
||||
> 2 * NATIVE_CHECKPOINT_INSPECTION_TOTAL_MAX_BYTES) return null;
|
||||
const cipher = createLocalEncryptedCipher("read");
|
||||
const records = new Map(rows.map(row => [row.part, decryptPart(cipher, binding, row.part, row.material)] as const));
|
||||
return inspectUnclaimedBootstrapCheckpoint(records, identity);
|
||||
} catch { return null; }
|
||||
}
|
||||
|
||||
export async function restoreNativeRunnerCheckpoint(db: Db, binding: Binding, root: string, normalizedSessionId?: string) {
|
||||
const checkpoint = await loadNativeRunnerCheckpoint(db, binding);
|
||||
if (!checkpoint) return null;
|
||||
|
||||
@@ -1,7 +1,9 @@
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import type { AdapterExecutionTarget } from "@paperclipai/adapter-utils/execution-target";
|
||||
import type { RunnerIngressEndpoint } from "@paperclipai/adapter-utils/runner-connectivity";
|
||||
import { prepareNativeRemoteRunner, createNativeRunnerExecutionDriver } from "./native-runner-execution-driver.js";
|
||||
import pino from "pino";
|
||||
import { environmentRunnerIngressErrorData, JsonRpcCallError, PluginEnvironmentRunnerIngressError } from "@paperclipai/plugin-sdk";
|
||||
import { NativeRunnerIngressAcquisitionError, prepareNativeRemoteRunner, createNativeRunnerExecutionDriver } from "./native-runner-execution-driver.js";
|
||||
|
||||
const transport = vi.hoisted(() => ({
|
||||
register: vi.fn(),
|
||||
@@ -38,6 +40,24 @@ beforeEach(() => {
|
||||
});
|
||||
|
||||
describe("shared Runner execution driver", () => {
|
||||
it.each(["ingress", "preparation", "unknown"])("limits ingress evidence and public error serialization at %s", async phase => {
|
||||
const failure = { phase: "private_preview", httpStatus: 502 } as const;
|
||||
const raw = new JsonRpcCallError({ code: -32000, message: "fixture-private-provider-response",
|
||||
data: phase === "unknown" ? undefined : environmentRunnerIngressErrorData(new PluginEnvironmentRunnerIngressError(failure)) });
|
||||
const acquire = vi.fn(async () => { throw raw; });
|
||||
const driver = createNativeRunnerExecutionDriver({ binding, runnerIngressAuthorized: true,
|
||||
target: { ...sandbox(), getRunnerIngressEndpoint: acquire } as AdapterExecutionTarget,
|
||||
prepareRemote: async () => { if (phase === "preparation") throw raw; },
|
||||
});
|
||||
const error = await driver.controlPlaneRegistration!(authority).catch(error => error);
|
||||
if (phase === "ingress") {
|
||||
expect(error).toBeInstanceOf(NativeRunnerIngressAcquisitionError);
|
||||
expect(error.failure).toEqual(failure);
|
||||
expect(JSON.stringify(pino.stdSerializers.err(error))).not.toContain("fixture-private");
|
||||
} else expect(error).toBe(raw);
|
||||
expect(transport.connect).not.toHaveBeenCalled();
|
||||
expect(transport.register).not.toHaveBeenCalled();
|
||||
});
|
||||
it("installs both staged checkpoint writers under the business lease without acquiring endpoints", async () => {
|
||||
const prepare = vi.fn();
|
||||
const persistence = { db: {} as never, bootId: "boot-c", authorityRunId: "run-c" };
|
||||
|
||||
@@ -7,6 +7,14 @@ import { connectRunnerPrpIngress } from "../../realtime/runner-prp-outbound.js";
|
||||
import type { NativeRunTrace } from "./native-run-trace.js";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { nativeRunnerCheckpointRegistration } from "./native-runner-checkpoint.js";
|
||||
import { readEnvironmentRunnerIngressFailure, type PluginEnvironmentRunnerIngressFailure } from "@paperclipai/plugin-sdk";
|
||||
|
||||
export class NativeRunnerIngressAcquisitionError extends Error {
|
||||
constructor(readonly failure: PluginEnvironmentRunnerIngressFailure) {
|
||||
super(`Runner connection endpoint could not be acquired (HTTP ${failure.httpStatus}).`);
|
||||
this.name = "NativeRunnerIngressAcquisitionError";
|
||||
}
|
||||
}
|
||||
|
||||
type ConnectionRegistration = NonNullable<RunnerExecutionDriver["controlPlaneRegistration"]>;
|
||||
type RunnerIdentity = Parameters<ConnectionRegistration>[1];
|
||||
@@ -177,21 +185,25 @@ export function createNativeRunnerExecutionDriver(input: NativeRunnerExecutionDr
|
||||
);
|
||||
// Provider endpoint acquisition happens only after runnerd is staged
|
||||
// and its listener capability has been verified.
|
||||
transport = await measureNativeRunnerSpan(
|
||||
input.trace,
|
||||
"runner.ingress.acquire",
|
||||
() =>
|
||||
resolvePaperclipRunnerTransport({
|
||||
target: input.target,
|
||||
runId:
|
||||
attachmentIdentity?.runId ??
|
||||
input.binding.runId,
|
||||
localConnectUrl: "ws://127.0.0.1/unused",
|
||||
runnerPublicUrl: input.runnerPublicUrl,
|
||||
runnerCaBundlePath: input.runnerCaBundlePath,
|
||||
runnerIngressAuthorized: true,
|
||||
}),
|
||||
);
|
||||
try {
|
||||
transport = await measureNativeRunnerSpan(
|
||||
input.trace,
|
||||
"runner.ingress.acquire",
|
||||
() =>
|
||||
resolvePaperclipRunnerTransport({
|
||||
target: input.target,
|
||||
runId: attachmentIdentity?.runId ?? input.binding.runId,
|
||||
localConnectUrl: "ws://127.0.0.1/unused",
|
||||
runnerPublicUrl: input.runnerPublicUrl,
|
||||
runnerCaBundlePath: input.runnerCaBundlePath,
|
||||
runnerIngressAuthorized: true,
|
||||
}),
|
||||
);
|
||||
} catch (error) {
|
||||
const failure = readEnvironmentRunnerIngressFailure(error);
|
||||
if (failure) throw new NativeRunnerIngressAcquisitionError(failure);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
const selectedAtMs = Date.now();
|
||||
|
||||
@@ -15335,7 +15335,7 @@ describe("runnerd provider runtime wiring", () => {
|
||||
runnerInstanceId: "runner-next", runnerExecutionTarget: remote as never }))
|
||||
.rejects.toThrow("runner_state_identity_mismatch: prior_owner_active");
|
||||
const projected = JSON.parse(await readFile(join(root, "control-plane/control-plane-state.json"), "utf8"));
|
||||
expect(projected).toMatchObject({ schema: "paperclip.runner.durable.control-plane-state.v1", commands: [] });
|
||||
expect(projected).toMatchObject({ schema: "paperclip.runner.durable.control-plane-state.v2", commands: [] });
|
||||
expect(projected.runAttachTemplate).toBeNull();
|
||||
const template = { provider: prior.provider, normalizedSessionId: identity.normalizedSessionId };
|
||||
controller.persistRunAttachTemplate(template);
|
||||
@@ -15508,7 +15508,7 @@ describe("runnerd provider runtime wiring", () => {
|
||||
if (["suspended", "backup", "codex", "file-projection", "legacy-file-no-checkpoint"].includes(scenario)) {
|
||||
await expect(resume).rejects.toBe(admitted);
|
||||
const restored = JSON.parse(await readFile(join(root, "control-plane/control-plane-state.json"), "utf8"));
|
||||
expect(restored).toMatchObject({ schema: "paperclip.runner.durable.control-plane-state.v1", identity });
|
||||
expect(restored).toMatchObject({ schema: "paperclip.runner.durable.control-plane-state.v2", identity });
|
||||
} else {
|
||||
const reason = ["missing-checkpoint", "foreign-task"].includes(scenario) ? "native_runner_checkpoint_missing"
|
||||
: ["checkpoint-binding-error", "file-checkpoint-binding-error"].includes(scenario) ? "native_runner_checkpoint_binding_mismatch"
|
||||
|
||||
@@ -31,7 +31,7 @@ import { type ProviderPackInstallation, verifyRemoteProviderPack as verifyRemote
|
||||
import { writeEnvironmentCleanup } from "../environment-cleanup-notifications.js";
|
||||
import { NATIVE_SESSION_EXECUTION_LEASE_TTL_MS, startNativeSessionExecutionLeaseRenewal, type NativeExecutionLeaseRenewal } from "./native-execution-lease.js";
|
||||
export { renewNativeSessionExecutionLease } from "./native-execution-lease.js";
|
||||
import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
|
||||
import { writeExecutionControl } from "../execution-control-notifications.js";
|
||||
import { notifyExecutionRunWork } from "../execution-control-notifications.js";
|
||||
import { createRunUsageRecorder } from "../usage-receipts.js";
|
||||
@@ -75,6 +75,7 @@ import { parseCodexCliVersion } from "./codex-runtime-compatibility.js";
|
||||
import { createNativeToolTrace, type NativeToolTrace } from "./native-tool-trace.js";
|
||||
import { createNativeProviderFailureObservation } from "./native-provider-failure.js";
|
||||
import { nativeStartupRetryExhaustedAction, nativeStartupRetryWait, planNativeStartupRetry } from "./native-startup-retry.js";
|
||||
import { admitUnclaimedNativeBootstrapRetry } from "./native-bootstrap-retry.js";
|
||||
import { createNativeDeliverableTextObservation } from "./native-deliverable-text.js";
|
||||
import { createNativeGitHubAccess, type NativeGitHubAccess } from "./native-github-access.js";
|
||||
import { resolveGitHubOperationCredentials } from "../github-operation-credentials.js";
|
||||
@@ -122,7 +123,7 @@ import {
|
||||
} from "node:fs";
|
||||
import { tmpdir, userInfo } from "node:os";
|
||||
import { DatabaseSync } from "node:sqlite";
|
||||
import { loadNativeRunnerCheckpointHeader, nativeRunnerCheckpointPersistence, nativeRunnerCheckpointRegistration, restoreNativeRunnerCheckpoint } from "./native-runner-checkpoint.js";
|
||||
import { loadNativeRunnerCheckpointHeader, loadNativeUnclaimedBootstrapCheckpoint, nativeRunnerCheckpointPersistence, nativeRunnerCheckpointRegistration, restoreNativeRunnerCheckpoint } from "./native-runner-checkpoint.js";
|
||||
import { REMOTE_CONTROLLER_LEASE_MS, REMOTE_RUNNER_RECONNECT_GRACE_MS } from "./native-runner-recovery-contract.js";
|
||||
import type { NativeTaskSessionPublicationContext } from "./native-task-session-publication.js";
|
||||
import { basename, dirname, join, posix, resolve } from "node:path";
|
||||
@@ -133,6 +134,7 @@ import type {
|
||||
import type { NativeFinalizationResult } from "@paperclipai/shared";
|
||||
import type {
|
||||
DurablePrpControlPlane,
|
||||
DurableRecoveryIdentity,
|
||||
HarnessRuntimeRequestResolution,
|
||||
NativeExecutionInput,
|
||||
NativeRuntimeContextSnapshot,
|
||||
@@ -179,7 +181,7 @@ import { createNativeSshCommandRunner } from "./native-ssh-command-runner.js";
|
||||
import { discoverRemoteExecutable, parseRemoteExecutableCandidate } from "./codex-model-fallback.js";
|
||||
export { parseRemoteExecutableCandidate } from "./codex-model-fallback.js";
|
||||
import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime";
|
||||
import { prepareNativeRemoteRunner, createNativeRunnerExecutionDriver, resolveRemoteRunnerTransportMode } from "./native-runner-execution-driver.js";
|
||||
import { NativeRunnerIngressAcquisitionError, prepareNativeRemoteRunner, createNativeRunnerExecutionDriver, resolveRemoteRunnerTransportMode } from "./native-runner-execution-driver.js";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import {
|
||||
and,
|
||||
@@ -484,8 +486,10 @@ const ACPX_RUNTIME_CHECKPOINT_EXCLUDED_ENTRIES = [
|
||||
"home/.npm",
|
||||
"data/pnpm/store",
|
||||
] as const;
|
||||
const RUNNERD_CONTROL_PLANE_STATE_SCHEMA =
|
||||
"paperclip.runner.durable.control-plane-state.v1";
|
||||
const RUNNERD_CONTROL_PLANE_STATE_SCHEMAS = new Set([
|
||||
"paperclip.runner.durable.control-plane-state.v1",
|
||||
"paperclip.runner.durable.control-plane-state.v2",
|
||||
]);
|
||||
const RUNNERD_STATE_SCHEMA = "paperclip.runner.durable.state.v1";
|
||||
const CODEX_PROVIDER_STATE_SCHEMA = "paperclip.runner.codex-provider-state.v1";
|
||||
const ACPX_PROVIDER_STATE_SCHEMA = "paperclip.runner.acpx-provider-state.v3";
|
||||
@@ -2953,7 +2957,7 @@ export async function verifyStoppedNativeSessionForContinuation(
|
||||
!Array.isArray(snapshot.control.committedEvents)) return null;
|
||||
// An incomplete provider launch can own a process that never emitted its
|
||||
// session identity. It cannot be certified from an earlier owner's receipt.
|
||||
if (snapshot.control.schema !== "paperclip.runner.durable.control-plane-state.v1" ||
|
||||
if (!RUNNERD_CONTROL_PLANE_STATE_SCHEMAS.has(String(snapshot.control.schema)) ||
|
||||
snapshot.runner.schema !== RUNNERD_STATE_SCHEMA ||
|
||||
snapshot.provider.schema !== (execution.provider.kind === "codex" ? "paperclip.runner.codex-provider-state.v1" : ACPX_PROVIDER_STATE_SCHEMA) ||
|
||||
!["turn_active", "prepared", "suspended"].includes(String(snapshot.provider.lifecycle)) ||
|
||||
@@ -5165,7 +5169,7 @@ function readSettledRunnerdProjection(root: string, identity: Record<string, str
|
||||
const answered = new Set((Array.isArray(control.commands) ? control.commands : []).map(record)
|
||||
.filter(command => command.type === "semantic_tool.result" && command.status === "completed")
|
||||
.map(command => record(command.payload).sourceEventId));
|
||||
if (control.schema !== RUNNERD_CONTROL_PLANE_STATE_SCHEMA || control.warmTransition != null ||
|
||||
if (!RUNNERD_CONTROL_PLANE_STATE_SCHEMAS.has(String(control.schema)) || control.warmTransition != null ||
|
||||
control.stagedRunAttachment != null || !isSettledRemoteWarmControlState(control, identity) ||
|
||||
(control.committedEvents as unknown[]).map(record).some(event =>
|
||||
(event.eventType === "semantic_tool.input" || event.eventType === "mcp_app.tool_input") &&
|
||||
@@ -5268,6 +5272,7 @@ async function migrateRunnerdStateRootForExecution(input: {
|
||||
onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise<void>;
|
||||
restartRecovery?: NativeRestartRecoveryClaim;
|
||||
runnerExecutionTarget?: AdapterExecutionTarget | null;
|
||||
bootstrapRetryLeaseOwner?: string;
|
||||
}): Promise<VerifiedWarmTransitionBinding | undefined> {
|
||||
const scoped = scopedRunnerdStateRoot(input.execution);
|
||||
await restoreWarmRunnerdController(input);
|
||||
@@ -5365,6 +5370,13 @@ async function migrateRunnerdStateRootForExecution(input: {
|
||||
return;
|
||||
}
|
||||
if (durableIdentityMatchesExecution(identity, input.execution)) {
|
||||
if (!input.restartRecovery && !input.retainedWarmOwnerRunId && input.bootstrapRetryLeaseOwner
|
||||
&& input.runnerExecutionTarget?.kind === "remote"
|
||||
&& runnerdAuthorityLifecycle(scoped, identity) === "absent"
|
||||
&& await admitUnclaimedNativeBootstrapRetry({ db: input.db, execution: input.execution,
|
||||
leaseOwner: input.bootstrapRetryLeaseOwner, preparedExecutionAdmission: input.preparedExecutionAdmission,
|
||||
localStateContents: readBoundedNativeFile(resolve(scoped, "control-plane", "control-plane-state.json"),
|
||||
8 * 1024 * 1024, "native_runner_checkpoint_invalid").toString("utf8") })) return;
|
||||
if (
|
||||
runnerdAuthorityLifecycleWithVerifiedBackup({
|
||||
root: scoped,
|
||||
@@ -5755,7 +5767,7 @@ function localProcessDefinitelyGone(value: unknown, group = false): boolean {
|
||||
function runnerdStateIsIncompleteBootstrap(state: Record<string, unknown>): boolean {
|
||||
const commands = Array.isArray(state.commands) ? state.commands.map(record) : [];
|
||||
const committedEvents = Array.isArray(state.committedEvents) ? state.committedEvents : [];
|
||||
return state.schema === RUNNERD_CONTROL_PLANE_STATE_SCHEMA &&
|
||||
return RUNNERD_CONTROL_PLANE_STATE_SCHEMAS.has(String(state.schema)) &&
|
||||
state.connectionCount === 0 && committedEvents.length === 0 &&
|
||||
commands.every(command => command.status === "pending" &&
|
||||
(command.type === "run.prepare" || command.type === "session.open"));
|
||||
@@ -5820,7 +5832,7 @@ function readRunnerdDurableIdentity(
|
||||
if (!staged) throw new Error("native_runner_staged_attachment_binding_mismatch");
|
||||
return { ...staged.sourceIdentity };
|
||||
}
|
||||
if (state.schema !== RUNNERD_CONTROL_PLANE_STATE_SCHEMA &&
|
||||
if (!RUNNERD_CONTROL_PLANE_STATE_SCHEMAS.has(String(state.schema)) &&
|
||||
!(allowExternal && state.schema === "paperclip.runner.external-control-plane-state.v1")) return null;
|
||||
return record(state.identity);
|
||||
} catch {
|
||||
@@ -7018,6 +7030,7 @@ export function nativeSessionFailureDisposition(
|
||||
sourceFailureCode?: ReturnType<typeof nativeSessionFailureSourceCode>,
|
||||
) {
|
||||
const permanentFailure =
|
||||
sourceFailureCode === "runner_ingress_access_rejected" ||
|
||||
sourceFailureCode === "native_provider_runtime_missing" ||
|
||||
sourceFailureCode === "native_provider_model_rejected" ||
|
||||
sourceFailureCode === "native_provider_approval_required" ||
|
||||
@@ -7101,6 +7114,8 @@ export function nativeSessionFailureSourceCode(
|
||||
| typeof NATIVE_CURRENT_WAKE_COMMENTS_UNREAD
|
||||
| typeof NATIVE_CURRENT_WAKE_COMMENTS_CHANGED_AFTER_READ
|
||||
| "provider_session_open_rejected"
|
||||
| "runner_ingress_access_rejected"
|
||||
| "runner_ingress_acquisition_failed"
|
||||
| "native_session_interrupted" {
|
||||
if (error instanceof NativePermissionDeclinedError) return "native_permission_declined";
|
||||
if (error instanceof NativeProviderStateIncompatibleError) return NATIVE_PROVIDER_STATE_INCOMPATIBLE;
|
||||
@@ -7120,6 +7135,10 @@ export function nativeSessionFailureSourceCode(
|
||||
return "native_event_replay_conflict";
|
||||
if (error instanceof NativeSessionCleanupQuarantinedError)
|
||||
return "native_session_cleanup_quarantined";
|
||||
if (error instanceof NativeRunnerIngressAcquisitionError) {
|
||||
return error.failure.httpStatus === 401 || error.failure.httpStatus === 403
|
||||
? "runner_ingress_access_rejected" : "runner_ingress_acquisition_failed";
|
||||
}
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
if (/classification=runtime_package_missing\)/.test(message))
|
||||
return "native_provider_runtime_missing";
|
||||
@@ -7217,14 +7236,22 @@ export async function nativeProviderRecoveryEvidence(input: {
|
||||
db: Db;
|
||||
runId: string;
|
||||
sourceFailureCode: ReturnType<typeof nativeSessionFailureSourceCode>;
|
||||
bootstrap?: {
|
||||
binding: { companyId: string; issueId: string; runId: string };
|
||||
identity: DurableRecoveryIdentity;
|
||||
transientIngressFailure: boolean;
|
||||
};
|
||||
}): Promise<{
|
||||
recoveryMode: NativeRecoveryMode;
|
||||
providerSessionEstablished: boolean;
|
||||
providerEventsExist: boolean;
|
||||
checkpointExists: boolean;
|
||||
unclaimedBootstrap?: { fingerprint: string };
|
||||
}> {
|
||||
const run = await input.db
|
||||
.select({ runnerProfileJson: heartbeatRuns.runnerProfileJson })
|
||||
.select({ runnerProfileJson: heartbeatRuns.runnerProfileJson, companyId: heartbeatRuns.companyId,
|
||||
nativeIssueId: heartbeatRuns.nativeIssueId, nativeSessionId: heartbeatRuns.nativeSessionId,
|
||||
runnerInstanceId: heartbeatRuns.runnerInstanceId })
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.id, input.runId))
|
||||
.limit(1)
|
||||
@@ -7251,6 +7278,12 @@ export async function nativeProviderRecoveryEvidence(input: {
|
||||
const providerEventsExist = durableEvents.some((event) =>
|
||||
PROVIDER_DURABLE_EVENT_TYPES.has(event.eventType),
|
||||
);
|
||||
const unclaimedBootstrap = !checkpointExists && !providerEventsExist && input.bootstrap
|
||||
&& input.bootstrap.binding.runId === input.runId && run?.companyId === input.bootstrap.binding.companyId
|
||||
&& run.nativeIssueId === input.bootstrap.binding.issueId && run.nativeSessionId === input.bootstrap.identity.normalizedSessionId
|
||||
&& run.runnerInstanceId === input.bootstrap.identity.runnerInstanceId
|
||||
? await loadNativeUnclaimedBootstrapCheckpoint(input.db, input.bootstrap.binding, input.bootstrap.identity)
|
||||
: null;
|
||||
if (
|
||||
checkpointExists &&
|
||||
providerSessionEstablished &&
|
||||
@@ -7278,12 +7311,14 @@ export async function nativeProviderRecoveryEvidence(input: {
|
||||
"native_runner_process_exited",
|
||||
"provider_session_open_rejected",
|
||||
]).has(input.sourceFailureCode);
|
||||
if (!checkpointExists && !providerEventsExist && definitelyPreSession) {
|
||||
if (!checkpointExists && !providerEventsExist && (definitelyPreSession
|
||||
|| unclaimedBootstrap && input.bootstrap?.transientIngressFailure)) {
|
||||
return {
|
||||
recoveryMode: "bootstrap_retry",
|
||||
providerSessionEstablished: false,
|
||||
providerEventsExist: false,
|
||||
checkpointExists: false,
|
||||
...(unclaimedBootstrap ? { unclaimedBootstrap: { fingerprint: unclaimedBootstrap.fingerprint } } : {}),
|
||||
};
|
||||
}
|
||||
return {
|
||||
@@ -7292,6 +7327,7 @@ export async function nativeProviderRecoveryEvidence(input: {
|
||||
providerSessionEstablished || providerEventsExist,
|
||||
providerEventsExist,
|
||||
checkpointExists,
|
||||
...(unclaimedBootstrap ? { unclaimedBootstrap: { fingerprint: unclaimedBootstrap.fingerprint } } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
@@ -8715,6 +8751,7 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
onLog: input.onLog,
|
||||
restartRecovery: input.restartRecovery,
|
||||
runnerExecutionTarget: input.runnerExecutionTarget,
|
||||
bootstrapRetryLeaseOwner: input.leaseOwner,
|
||||
});
|
||||
}
|
||||
const durableRunnerBinding = input.useRunnerd
|
||||
@@ -10449,10 +10486,22 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
: providerUsageLimitObserved
|
||||
? "native_provider_usage_limit"
|
||||
: classifiedFailureCode;
|
||||
const bootstrap = input.useRunnerd && input.runnerExecutionTarget?.kind === "remote"
|
||||
&& !input.restartRecovery && !retainedTransition && !remoteWarmRecovery
|
||||
? {
|
||||
binding: input.execution.binding,
|
||||
identity: { runnerInstanceId: effectiveRunnerInstanceId,
|
||||
environmentLeaseId: durableRunnerBinding?.environmentLeaseId ?? input.execution.binding.executionWorkspaceId,
|
||||
runId: input.execution.binding.runId, normalizedSessionId: nativeSessionKey(input.execution),
|
||||
turnId: `turn-${input.execution.binding.runId}`, itemId: `item-${input.execution.binding.runId}` },
|
||||
transientIngressFailure: sourceFailureCode === "runner_ingress_acquisition_failed" && error instanceof NativeRunnerIngressAcquisitionError
|
||||
&& (error.failure.httpStatus === 429 || error.failure.httpStatus >= 500),
|
||||
} : undefined;
|
||||
const recoveryEvidence = await nativeProviderRecoveryEvidence({
|
||||
db: input.db,
|
||||
runId: input.execution.binding.runId,
|
||||
sourceFailureCode,
|
||||
bootstrap,
|
||||
});
|
||||
const disposition = nativeSessionFailureDisposition(
|
||||
attempt,
|
||||
@@ -10473,6 +10522,9 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
const startupRetryExhausted =
|
||||
recoveryEvidence.recoveryMode === "bootstrap_retry" &&
|
||||
failureCode === "native_session_retry_exhausted";
|
||||
const startupExhaustedAction = sourceFailureCode === "runner_ingress_acquisition_failed"
|
||||
? `Paperclip could not acquire the computer's connection endpoint after ${attempt} attempts. Automatic retries are stopped. Inspect the computer provider's service status before retrying.`
|
||||
: null;
|
||||
const startupRetry =
|
||||
recoveryEvidence.recoveryMode === "bootstrap_retry" &&
|
||||
(phase === "retryable_failure" || startupRetryExhausted)
|
||||
@@ -10541,6 +10593,7 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
message,
|
||||
...(providerFailure ? { providerFailure } : {}),
|
||||
...(remoteRunnerRecovery ? { remoteRunnerRecovery } : {}),
|
||||
...(error instanceof NativeRunnerIngressAcquisitionError ? { runnerIngressFailure: error.failure } : {}),
|
||||
originalFailureCode:
|
||||
error instanceof NativeProviderTerminalFailure
|
||||
? error.providerCode
|
||||
@@ -10565,6 +10618,8 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
? NATIVE_CLEANUP_OPERATOR_RECOVERY_MESSAGE
|
||||
: sourceFailureCode === "runner_remote_provider_verification_unavailable"
|
||||
? "Provider files could not be verified after two read attempts. Inspect the retained workspace and command transport before retrying the original run."
|
||||
: sourceFailureCode === "runner_ingress_access_rejected"
|
||||
? "The computer provider rejected access. Restore the environment's credentials or permissions before retrying. Automatic retries are stopped."
|
||||
: sourceFailureCode === "native_provider_terminal_failed"
|
||||
? "The provider session is permanently unusable. Verify stopped execution, completed actions, and task context before starting a linked continuation."
|
||||
: sourceFailureCode === NATIVE_PROVIDER_STATE_INCOMPATIBLE
|
||||
@@ -10578,7 +10633,7 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
: ownershipUnverified
|
||||
? "Inspect the retained runner's executable and authenticated connection. Do not replace its provider until ownership is safely resolved."
|
||||
: startupRetryExhausted
|
||||
? nativeStartupRetryExhaustedAction(attempt, providerCooldownAt)
|
||||
? startupExhaustedAction ?? nativeStartupRetryExhaustedAction(attempt, providerCooldownAt)
|
||||
: exhausted
|
||||
? "Inspect the persisted native session after its bounded resume budget was exhausted."
|
||||
: recoveryEvidence.recoveryMode ===
|
||||
@@ -10640,6 +10695,8 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
),
|
||||
eq(nativeRunFinalizations.leaseOwner, leaseOwner),
|
||||
eq(nativeRunFinalizations.attempt, attempt),
|
||||
eq(nativeRunFinalizations.controllerGeneration, controllerGeneration),
|
||||
isNull(nativeRunFinalizations.resultId),
|
||||
eq(nativeRunFinalizations.controllerBootId, controller.bootId),
|
||||
eq(nativeRunFinalizations.controllerPid, controller.pid),
|
||||
eq(
|
||||
@@ -10653,10 +10710,27 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!updated) throw new Error("native_session_lease_lost");
|
||||
const [runBeforeWrite] = await tx
|
||||
.select({ status: heartbeatRuns.status })
|
||||
.select({ status: heartbeatRuns.status, resultJson: heartbeatRuns.resultJson,
|
||||
runnerInstanceId: heartbeatRuns.runnerInstanceId, nativeSessionId: heartbeatRuns.nativeSessionId,
|
||||
costAccountedAt: heartbeatRuns.costAccountedAt })
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.id, input.execution.binding.runId))
|
||||
.for("update");
|
||||
if (recoveryEvidence.unclaimedBootstrap && bootstrap) {
|
||||
const currentEvidence = await nativeProviderRecoveryEvidence({ db: tx as unknown as Db,
|
||||
runId: input.execution.binding.runId, sourceFailureCode, bootstrap });
|
||||
if (!runBeforeWrite || ["cancelled", "succeeded"].includes(runBeforeWrite.status)
|
||||
|| Object.keys(record(runBeforeWrite.resultJson?.nativeCancellation)).length > 0
|
||||
|| runBeforeWrite.runnerInstanceId !== bootstrap.identity.runnerInstanceId
|
||||
|| runBeforeWrite.nativeSessionId !== bootstrap.identity.normalizedSessionId
|
||||
|| currentEvidence.unclaimedBootstrap?.fingerprint !== recoveryEvidence.unclaimedBootstrap.fingerprint
|
||||
|| currentEvidence.recoveryMode !== recoveryEvidence.recoveryMode) {
|
||||
throw new Error("native_runner_bootstrap_evidence_changed");
|
||||
}
|
||||
}
|
||||
const unusedTerminal = !!recoveryEvidence.unclaimedBootstrap && phase === "terminal_failure"
|
||||
&& !runBeforeWrite?.costAccountedAt;
|
||||
if (unusedTerminal) await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.execution.binding.companyId);
|
||||
if (!ownershipUnverified) await notifyExternalAgentSettlement(tx, input.execution.binding);
|
||||
if (!ownershipUnverified) await notifyInstructionCollection(tx, { companyId: input.execution.binding.companyId, runId: input.execution.binding.runId });
|
||||
if (!ownershipUnverified) await notifyCancelledWorkspacePreservation(tx, { companyId: input.execution.binding.companyId, runId: input.execution.binding.runId });
|
||||
@@ -10675,6 +10749,10 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
: {}),
|
||||
nativePhase: phase,
|
||||
nativePhaseUpdatedAt: now,
|
||||
...(unusedTerminal ? {
|
||||
usageJson: sql`coalesce(${heartbeatRuns.usageJson}, '{}'::jsonb) || '{"accountingProviderWorkStarted":false}'::jsonb`,
|
||||
costAccountingPending: true,
|
||||
} : {}),
|
||||
error: message,
|
||||
resultJson: sql`(coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) - 'remoteRunnerRecovery')
|
||||
|| ${JSON.stringify(remoteRunnerRecovery ? { remoteRunnerRecovery } : {})}::jsonb
|
||||
@@ -10759,6 +10837,8 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
? "Approval required. Review the operation and update the agent's permission setting before retrying. This runner has no interactive approval handler."
|
||||
: sourceFailureCode === "native_session_cleanup_quarantined"
|
||||
? NATIVE_CLEANUP_OPERATOR_RECOVERY_MESSAGE
|
||||
: sourceFailureCode === "runner_ingress_access_rejected"
|
||||
? "The computer provider rejected access. Restore the environment's credentials or permissions before retrying. Automatic retries are stopped."
|
||||
: sourceFailureCode === "native_provider_terminal_failed"
|
||||
? "Verify that the failed provider stopped and reconcile its action outcomes. A linked continuation can proceed only after these checks succeed."
|
||||
: sourceFailureCode === NATIVE_PROVIDER_STATE_INCOMPATIBLE
|
||||
@@ -10772,7 +10852,7 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
: ownershipUnverified
|
||||
? "Resolve the retained runner's authentication or executable compatibility before an explicit recovery; do not blindly restart, cancel, or replace its provider session."
|
||||
: startupRetryExhausted
|
||||
? nativeStartupRetryExhaustedAction(attempt, providerCooldownAt)
|
||||
? startupExhaustedAction ?? nativeStartupRetryExhaustedAction(attempt, providerCooldownAt)
|
||||
: exhausted
|
||||
? "Inspect the provider trace and explicitly choose a replacement run or provider configuration; automatic provider work is stopped."
|
||||
: recoveryEvidence.recoveryMode === "bootstrap_retry"
|
||||
|
||||
@@ -1,5 +1,9 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { mkdtempSync, rmSync } from "node:fs";
|
||||
import { createServer } from "node:http";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
agents,
|
||||
@@ -11,6 +15,7 @@ import {
|
||||
issueRecoveryActions,
|
||||
issues,
|
||||
nativeRunFinalizations,
|
||||
nativeRunnerCheckpoints,
|
||||
} from "@paperclipai/db";
|
||||
import type { NativeExecutionInputV1 } from "@paperclipai/paperclip-runner";
|
||||
import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
|
||||
@@ -27,6 +32,12 @@ import { executePaperclipNativeSession } from "./native-session-executor.js";
|
||||
import { buildNativeCompletionContract } from "./completion-contracts.js";
|
||||
import { prepareNativeHeartbeatRun } from "./prepare-native-run.js";
|
||||
import { claimNativeSessionResumptions } from "./native-finalization-reconciler.js";
|
||||
import { DurablePrpControlPlane } from "../../vendor/paperclip-runner/index.js";
|
||||
import { createNativeRunnerExecutionDriver } from "./native-runner-execution-driver.js";
|
||||
import { currentNativeControllerIdentity } from "./native-restart-recovery.js";
|
||||
import { loadNativeRunnerCheckpoint } from "./native-runner-checkpoint.js";
|
||||
import { environmentRunnerIngressErrorData, JsonRpcCallError, PluginEnvironmentRunnerIngressError } from "@paperclipai/plugin-sdk";
|
||||
import { accountRunCost } from "../run-cost-accounting.js";
|
||||
|
||||
type SessionOpenStatus = "failed" | "rejected" | "indeterminate";
|
||||
|
||||
@@ -272,6 +283,150 @@ describe("native provider startup retry", () => {
|
||||
});
|
||||
}
|
||||
|
||||
it.each([
|
||||
{ name: "transient 502", status: 502, retry: true },
|
||||
{ name: "credential 401", status: 401, retry: false },
|
||||
{ name: "permission 403", status: 403, retry: false },
|
||||
{ name: "invalid request 400", status: 400, retry: false },
|
||||
{ name: "untyped HTML 502", status: 502, retry: false, raw: true },
|
||||
{ name: "previously issued credential", status: 502, retry: false, checkpoint: "issued" },
|
||||
{ name: "legacy checkpoint", status: 502, retry: false, checkpoint: "legacy" },
|
||||
{ name: "truncated checkpoint", status: 502, retry: false, checkpoint: "truncated" },
|
||||
{ name: "provider progress", status: 502, retry: false, checkpoint: "progress" },
|
||||
{ name: "checkpoint change during settlement", status: 502, retry: false, race: "checkpoint" },
|
||||
{ name: "Stop during settlement", status: 502, retry: false, race: "stop" },
|
||||
{ name: "new controller claim during settlement", status: 502, retry: false, race: "claim" },
|
||||
{ name: "exhausted transient 502", status: 502, retry: false, finalAttempt: true },
|
||||
])("settles ingress failure safely: $name", async options => {
|
||||
const seeded = await seedRun(await createAgent());
|
||||
if (options.finalAttempt) await db.update(nativeRunFinalizations).set({ attempt: 2 }).where(eq(nativeRunFinalizations.runId, seeded.runId));
|
||||
const root = mkdtempSync(join(tmpdir(), "native-ingress-bootstrap-"));
|
||||
const server = createServer((_request, response) => {
|
||||
response.writeHead(options.status, { "content-type": "text/html" });
|
||||
response.end(`<html><title>${options.status} Bad Gateway</title></html>`);
|
||||
});
|
||||
await new Promise<void>(resolve => server.listen(0, "127.0.0.1", resolve));
|
||||
const address = server.address();
|
||||
if (!address || typeof address === "string") throw new Error("test server missing address");
|
||||
const identity = {
|
||||
runnerInstanceId: seeded.runnerInstanceId,
|
||||
environmentLeaseId: seeded.execution.binding.executionWorkspaceId,
|
||||
runId: seeded.runId,
|
||||
normalizedSessionId: seeded.execution.session.normalizedSessionId!,
|
||||
turnId: `turn-${seeded.runId}`,
|
||||
itemId: `item-${seeded.runId}`,
|
||||
};
|
||||
const core = new DurablePrpControlPlane({ stateDirectory: root, identity,
|
||||
expectedRunnerVersion: "0.3.0", expectedRunnerDigest: `sha256:${"a".repeat(64)}` });
|
||||
const prepare = { observeCredentials: true, authorizedTools: {}, provider: { provider: "codex" } };
|
||||
core.persistRunAttachTemplate(prepare);
|
||||
core.queueCommand("run.prepare", prepare);
|
||||
core.queueCommand("session.open", { reuse: "same_session" });
|
||||
if (options.checkpoint === "issued") {
|
||||
core.issueBootstrapTicket();
|
||||
core.store.state.tickets = {};
|
||||
}
|
||||
if (options.checkpoint === "legacy") {
|
||||
core.store.state.schema = "paperclip.runner.durable.control-plane-state.v1";
|
||||
delete core.store.state.bootstrapHistory;
|
||||
}
|
||||
const launch = vi.fn();
|
||||
let settlementArmed = false;
|
||||
let settlementTransactions = 0;
|
||||
const transaction = db.transaction.bind(db);
|
||||
const settlementRace = options.race ? vi.spyOn(db, "transaction").mockImplementation(async (...args) => {
|
||||
if (settlementArmed && ++settlementTransactions === 2) {
|
||||
settlementArmed = false;
|
||||
if (options.race === "checkpoint") { core.issueBootstrapTicket(); await core.flushCheckpoint(); }
|
||||
if (options.race === "stop") await db.update(heartbeatRuns).set({ resultJson: {
|
||||
nativeCancellation: { schema: "paperclip.native-cancellation.v1", companyId, issueId: seeded.issueId,
|
||||
runId: seeded.runId, scope: "run", intentAuditId: randomUUID(), dispatchState: "pending" },
|
||||
} }).where(eq(heartbeatRuns.id, seeded.runId));
|
||||
if (options.race === "claim") await db.update(nativeRunFinalizations).set({
|
||||
controllerGeneration: sql`${nativeRunFinalizations.controllerGeneration} + 1`, leaseOwner: "successor-controller",
|
||||
}).where(eq(nativeRunFinalizations.runId, seeded.runId));
|
||||
}
|
||||
return transaction(...args);
|
||||
}) : undefined;
|
||||
const getRunnerIngressEndpoint = vi.fn(async () => {
|
||||
const response = await fetch(`http://127.0.0.1:${address.port}/private-preview`);
|
||||
if (options.checkpoint === "truncated") await db.delete(nativeRunnerCheckpoints).where(and(
|
||||
eq(nativeRunnerCheckpoints.companyId, companyId), eq(nativeRunnerCheckpoints.runId, seeded.runId),
|
||||
eq(nativeRunnerCheckpoints.part, "command:2")));
|
||||
if (options.checkpoint === "progress") await db.insert(heartbeatRunEvents).values({
|
||||
companyId, runId: seeded.runId, agentId: seeded.agentId, seq: 900, eventType: "turn.started", stream: "system", level: "info",
|
||||
message: "Provider activity exists", payload: {},
|
||||
});
|
||||
if (!response.ok) {
|
||||
const cause = new Error(await response.text());
|
||||
if (options.raw) throw cause;
|
||||
const error = new PluginEnvironmentRunnerIngressError({ phase: "private_preview", httpStatus: response.status });
|
||||
settlementArmed = true;
|
||||
throw new JsonRpcCallError({ code: -32000, message: error.message, data: environmentRunnerIngressErrorData(error) });
|
||||
}
|
||||
throw new Error("expected the controlled ingress rejection");
|
||||
});
|
||||
provider.execute.mockReset().mockImplementation(async () => {
|
||||
const controller = await currentNativeControllerIdentity();
|
||||
const driver = createNativeRunnerExecutionDriver({
|
||||
target: { kind: "remote", transport: "sandbox", remoteCwd: "/workspace", leaseId: randomUUID(),
|
||||
effectiveCapabilities: { runnerWebSocketIngress: true }, getRunnerIngressEndpoint } as never,
|
||||
binding: seeded.execution.binding, runnerIngressAuthorized: true, prepareRemote: async () => {},
|
||||
controllerPersistence: { db, bootId: controller.bootId },
|
||||
processStarted: new Promise<void>(() => {}), process: { runnerProcessLauncher: launch },
|
||||
});
|
||||
await driver.controlPlaneRegistration!(core);
|
||||
throw new Error("ingress failure must precede process launch");
|
||||
});
|
||||
try {
|
||||
await expect(executePaperclipNativeSession({ db, execution: seeded.execution,
|
||||
runnerInstanceId: seeded.runnerInstanceId, useRunnerd: true,
|
||||
runnerExecutionTarget: { kind: "remote", transport: "sandbox", remoteCwd: "/workspace", runner: {} } as never,
|
||||
backend: { descriptor: async () => { throw new Error("unexpected descriptor"); },
|
||||
openSession: async () => { throw new Error("unexpected provider session"); } },
|
||||
})).rejects.toThrow(options.race === "claim" ? "native_session_lease_lost" : options.race
|
||||
? "native_runner_bootstrap_evidence_changed" : options.raw ? "502 Bad Gateway" : `HTTP ${options.status}`);
|
||||
expect(getRunnerIngressEndpoint).toHaveBeenCalledOnce();
|
||||
expect(launch).not.toHaveBeenCalled();
|
||||
if (options.checkpoint !== "truncated") {
|
||||
const saved = await loadNativeRunnerCheckpoint(db, seeded.execution.binding);
|
||||
expect(saved?.recoveryEvidence.authenticatedConnectionCount).toBe(0);
|
||||
expect(saved?.records.size).toBe(4);
|
||||
}
|
||||
const state = await read(seeded);
|
||||
if (options.race) {
|
||||
expect(state.coordinator.phase).not.toBe("retryable_failure");
|
||||
expect(state.coordinator.phase).not.toBe("terminal_failure");
|
||||
expect(state.coordinator.leaseOwner).not.toBeNull();
|
||||
expect(state.issue.executionRunId).toBe(seeded.runId);
|
||||
expect(state.run.status).toBe("running");
|
||||
expect(state.run.costAccountedAt).toBeNull();
|
||||
expect(state.run.usageJson?.accountingProviderWorkStarted).not.toBe(false);
|
||||
return;
|
||||
}
|
||||
expect(state.coordinator).toMatchObject({ phase: options.retry ? "retryable_failure" : "terminal_failure", attempt: options.finalAttempt ? 3 : 1,
|
||||
failureDetail: { recoveryMode: options.retry || options.finalAttempt ? "bootstrap_retry" : "ambiguous_state",
|
||||
providerSessionEstablished: options.checkpoint === "progress", providerEventsExist: options.checkpoint === "progress", checkpointExists: false } });
|
||||
if (options.retry) expect(state.coordinator.nextAttemptAt).toBeInstanceOf(Date);
|
||||
else expect(state.coordinator.nextAttemptAt).toBeNull();
|
||||
expect(state.issue.status).toBe(options.retry ? "in_progress" : "blocked");
|
||||
expect(state.run.costAccountedAt).toBeNull();
|
||||
expect(state.coordinator.resultId).toBeNull();
|
||||
if (options.retry || options.checkpoint) {
|
||||
expect(state.run.usageJson?.accountingProviderWorkStarted).not.toBe(false);
|
||||
} else {
|
||||
expect(state.run.usageJson?.accountingProviderWorkStarted).toBe(false);
|
||||
expect(await accountRunCost(db, seeded.runId)).toBe(true);
|
||||
expect((await read(seeded)).run.costAccountedAt).toBeInstanceOf(Date);
|
||||
}
|
||||
} finally {
|
||||
settlementRace?.mockRestore();
|
||||
await core.stop();
|
||||
await new Promise<void>((resolve, reject) => server.close(error => error ? reject(error) : resolve()));
|
||||
rmSync(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("retries a rejected session.open with growing backoff instead of blocking the task", async () => {
|
||||
const seeded = await seedRun(await createAgent());
|
||||
|
||||
|
||||
+2
@@ -96,6 +96,8 @@ export const providerFailureFromError = runner.providerFailureFromError;
|
||||
export const decodeDurablePrpCheckpoint = runner.decodeDurablePrpCheckpoint;
|
||||
export const captureLegacyDurablePrpCheckpoint = runner.captureLegacyDurablePrpCheckpoint;
|
||||
export const inspectDurablePrpCheckpointHeader = runner.inspectDurablePrpCheckpointHeader;
|
||||
export const inspectUnclaimedBootstrapCheckpoint = runner.inspectUnclaimedBootstrapCheckpoint;
|
||||
export const inspectUnclaimedBootstrapState = runner.inspectUnclaimedBootstrapState;
|
||||
export const RUNNER_PRP_PROTOCOL_MIN = runner.RUNNER_PRP_PROTOCOL_MIN;
|
||||
export const RUNNER_PRP_PROTOCOL_MAX = runner.RUNNER_PRP_PROTOCOL_MAX;
|
||||
export const runnerCodexDynamicToolsFit = runner.runnerCodexDynamicToolsFit;
|
||||
|
||||
@@ -32,6 +32,7 @@ export const releaseMaterializedNativeRuntimeSkills: RunnerTestingModule["releas
|
||||
runnerTesting.releaseMaterializedNativeRuntimeSkills;
|
||||
export const resolveRunnerdCodexSkillInputs: RunnerTestingModule["resolveRunnerdCodexSkillInputs"] =
|
||||
runnerTesting.resolveRunnerdCodexSkillInputs;
|
||||
export const spawnRunner: RunnerTestingModule["spawnRunner"] = runnerTesting.spawnRunner;
|
||||
|
||||
export const startOpenCodeMcpBridge:
|
||||
RunnerTestingModule["startOpenCodeMcpBridge"] =
|
||||
|
||||
Reference in new issue
Block a user