mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-21 08:22:20 +02:00
fix(runner): persist warm Daytona workspaces (#12901)
## Thinking Path > - Paperclip manages AI agent work and the execution state for each task. > - Remote agents run in sandbox environments such as Daytona. > - Daytona keeps files while a sandbox is stopped, but deletion removes those files. > - Runner Codex did not copy successful remote workspace changes back to the host workspace. > - A warm sandbox could therefore hide data loss until Daytona replaced or deleted the sandbox. > - This pull request makes the host workspace durable after every successful turn and keeps verified reusable sandboxes warm. > - The benefit is reliable multi-turn work across warm reuse, restart, stop, and sandbox replacement. ## Linked Issues or Issue Description **What happened?** A successful native Codex turn in Daytona could leave workspace changes only in the remote sandbox. A later warm turn appeared to work because it reused that filesystem. A replacement sandbox could start from stale host data and lose the successful changes. **Expected behavior** Paperclip must merge each successful remote turn into the authoritative host workspace before it completes the run. A verified warm lease may reuse its remote files. A replacement lease must reconstruct the exact durable workspace seed. **Steps to reproduce** 1. Run Codex in a reusable Daytona environment. 2. Write a file during one successful turn. 3. Replace the Daytona sandbox before the next turn. 4. Observe that the next turn can start without the prior file on the unpatched code. Related remote workspace foundation: #10070. ## What Changed - Added explicit `host_current`, `durable_seed`, and `adopt_remote` workspace preparation modes. - Added atomic, versioned native workspace descriptors and seed archives under `PAPERCLIP_HOME`. - Added real native sandbox export and three-way host merge before terminal result completion. - Added workspace-only recovery after a proposed result. Recovery does not submit another provider turn or consume the provider retry budget. - Added fail-closed handling when a sandbox with unexported changes is gone. - Kept healthy reusable Daytona sandboxes started for legacy Codex and Runner Codex. - Kept the Runner Codex process and provider session across verified warm turns. - Added the paid `daytona-warm-continuity` browser suite. It contains exactly the legacy Codex and Runner Codex cells. Each cell performs three measured turns. - Documented `pnpm test:e2e:runner -- --suite daytona-warm-continuity`. No package script was added. - Added no database migration. The metadata format is backward compatible and idempotent. ## Verification - `pnpm typecheck` - `pnpm test:e2e:runner:unit` — 114 passed - Native workspace, finalizer, session, and environment tests — 232 passed - Daytona provider tests — 150 passed - Workspace staging and merge tests — 98 passed - Runner transport tests — 63 passed - Legacy Codex restore tests — 5 passed - Rust format and compile checks pass through root typecheck - The paid Daytona suite was not run locally because the required Daytona, OpenAI, and immutable image credentials are not present. ## Risks - The main risk is an incorrect workspace identity or merge after a crash. Durable descriptors bind the run, workspace, lease, provider lease, local root, remote root, and baseline digest. Ambiguous evidence fails closed. - The host merge may conflict with concurrent host edits. The existing three-way merge and exclusion rules handle this case and surface failures. - A deleted sandbox cannot recover unexported bytes. Paperclip now blocks with `workspace_sync_out_unrecoverable` instead of reporting success or rerunning the provider. - There is no database migration. Descriptor writes and recovery are atomic and idempotent. ## Model Used OpenAI Codex with GPT-5. The run used agentic reasoning, repository inspection, code execution, test execution, Git, and GitHub CLI tools. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge
This commit is contained in:
Dotta authored and GitHub committed
1 parent
be407f3456
commit
1dceee9a4e
47 files changed
+4296
-359
No files matched your search
@@ -561,6 +561,18 @@ describeEmbeddedPostgres("environmentRuntimeService", () => {
|
||||
mode: "shared_workspace",
|
||||
},
|
||||
});
|
||||
const workspaceSyncStamp = {
|
||||
schema: "paperclip.native-workspace-stamp/v1",
|
||||
workspaceId: seeded.executionWorkspaceId,
|
||||
providerLeaseId: "sandbox-exact-resume",
|
||||
remoteCwd: "/workspace",
|
||||
hostSha256: "a".repeat(64),
|
||||
finalizedRunId: seeded.runId,
|
||||
};
|
||||
await environmentService(db).updateLeaseMetadata(first.lease.id, {
|
||||
...(first.lease.metadata ?? {}),
|
||||
nativeWorkspaceSync: workspaceSyncStamp,
|
||||
});
|
||||
await runtimeWithPlugin.releaseRunLeases(
|
||||
seeded.runId,
|
||||
"released",
|
||||
@@ -592,9 +604,17 @@ describeEmbeddedPostgres("environmentRuntimeService", () => {
|
||||
|
||||
expect(first.lease.metadata?.sandboxLeaseAcquisition).toEqual({ outcome: "created" });
|
||||
expect(acquired.lease.providerLeaseId).toBe("sandbox-exact-resume");
|
||||
expect(acquired.lease.metadata?.sandboxLeaseAcquisition).toEqual({ outcome: "resumed" });
|
||||
expect(workerManager.call.mock.calls.filter((call) => call[1] === "environmentAcquireLease"))
|
||||
.toHaveLength(1);
|
||||
expect(acquired.lease.metadata?.sandboxLeaseAcquisition).toEqual({
|
||||
outcome: "resumed",
|
||||
});
|
||||
expect(acquired.lease.metadata?.nativeWorkspaceSync).toEqual(
|
||||
workspaceSyncStamp,
|
||||
);
|
||||
expect(
|
||||
workerManager.call.mock.calls.filter(
|
||||
(call) => call[1] === "environmentAcquireLease",
|
||||
),
|
||||
).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("destroys a disposable paperclip_runner sandbox after the turn", async () => {
|
||||
|
||||
@@ -1,5 +1,9 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { resolveNativeSandboxLifecycle } from "../services/heartbeat.js";
|
||||
import {
|
||||
providerResourceDispositionForTerminalRun,
|
||||
resolveNativeSandboxLifecycle,
|
||||
resolveReusableSandboxLifecycle,
|
||||
} from "../services/heartbeat.js";
|
||||
|
||||
const reusableSandbox = {
|
||||
kind: "remote" as const,
|
||||
@@ -21,6 +25,19 @@ describe("paperclip_runner sandbox lifecycle", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it("keeps the same warm reusable sandbox for a legacy adapter", () => {
|
||||
expect(
|
||||
resolveReusableSandboxLifecycle({
|
||||
lifecyclePolicy: { mode: "warm", idleTimeoutMs: 300_000 },
|
||||
target: reusableSandbox,
|
||||
}),
|
||||
).toEqual({
|
||||
runnerProcess: "warm",
|
||||
sandboxResource: "keep_running",
|
||||
failoverBackup: "verified",
|
||||
});
|
||||
});
|
||||
|
||||
it("stops and reuses a per-turn reusable sandbox", () => {
|
||||
expect(resolveNativeSandboxLifecycle({
|
||||
adapterType: "paperclip_runner",
|
||||
@@ -71,4 +88,22 @@ describe("paperclip_runner sandbox lifecycle", () => {
|
||||
target: { kind: "local" },
|
||||
})).toBeNull();
|
||||
});
|
||||
|
||||
it("keeps a warm sandbox only after a successful turn", () => {
|
||||
expect(
|
||||
providerResourceDispositionForTerminalRun("keep_running", "succeeded"),
|
||||
).toBe("keep_running");
|
||||
expect(
|
||||
providerResourceDispositionForTerminalRun("keep_running", "failed"),
|
||||
).toBe("stop_and_retain");
|
||||
expect(
|
||||
providerResourceDispositionForTerminalRun("keep_running", "cancelled"),
|
||||
).toBe("stop_and_retain");
|
||||
expect(
|
||||
providerResourceDispositionForTerminalRun("keep_running", "timed_out"),
|
||||
).toBe("stop_and_retain");
|
||||
expect(providerResourceDispositionForTerminalRun("destroy", "failed")).toBe(
|
||||
"destroy",
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,297 @@
|
||||
import { mkdtemp, readdir, rm } from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
directorySnapshotSha256,
|
||||
serializeDirectorySnapshot,
|
||||
} from "@paperclipai/adapter-utils/workspace-restore-merge";
|
||||
|
||||
import {
|
||||
classifyNativeWorkspaceInbound,
|
||||
nativeWorkspaceSyncInternals,
|
||||
readNativeWorkspaceSyncReference,
|
||||
resumeNativeWorkspaceSync,
|
||||
} from "../services/native-runtime/native-workspace-sync.js";
|
||||
|
||||
const digest = "a".repeat(64);
|
||||
|
||||
describe("native workspace sync durable metadata", () => {
|
||||
const originalPaperclipHome = process.env.PAPERCLIP_HOME;
|
||||
const originalPaperclipInstanceId = process.env.PAPERCLIP_INSTANCE_ID;
|
||||
const cleanupDirs: string[] = [];
|
||||
|
||||
afterEach(async () => {
|
||||
if (originalPaperclipHome === undefined) delete process.env.PAPERCLIP_HOME;
|
||||
else process.env.PAPERCLIP_HOME = originalPaperclipHome;
|
||||
if (originalPaperclipInstanceId === undefined)
|
||||
delete process.env.PAPERCLIP_INSTANCE_ID;
|
||||
else process.env.PAPERCLIP_INSTANCE_ID = originalPaperclipInstanceId;
|
||||
await Promise.all(
|
||||
cleanupDirs
|
||||
.splice(0)
|
||||
.map((directory) => rm(directory, { recursive: true, force: true })),
|
||||
);
|
||||
});
|
||||
|
||||
it("classifies fresh, warm, replacement, and same-run recovery inputs", () => {
|
||||
expect(
|
||||
classifyNativeWorkspaceInbound({
|
||||
kind: "new_run",
|
||||
acquisition: "created",
|
||||
hasPriorStamp: false,
|
||||
}),
|
||||
).toBe("host_current");
|
||||
expect(
|
||||
classifyNativeWorkspaceInbound({
|
||||
kind: "new_run",
|
||||
acquisition: "resumed",
|
||||
hasPriorStamp: true,
|
||||
}),
|
||||
).toBe("adopt_remote");
|
||||
expect(
|
||||
classifyNativeWorkspaceInbound({
|
||||
kind: "new_run",
|
||||
acquisition: "resumed",
|
||||
hasPriorStamp: false,
|
||||
}),
|
||||
).toBe("host_current");
|
||||
expect(
|
||||
classifyNativeWorkspaceInbound({
|
||||
kind: "new_run",
|
||||
acquisition: "replacement",
|
||||
hasPriorStamp: true,
|
||||
}),
|
||||
).toBe("host_current");
|
||||
expect(
|
||||
classifyNativeWorkspaceInbound({
|
||||
kind: "existing_run",
|
||||
restartRecovery: true,
|
||||
sameProviderLease: true,
|
||||
}),
|
||||
).toBe("adopt_remote");
|
||||
expect(
|
||||
classifyNativeWorkspaceInbound({
|
||||
kind: "existing_run",
|
||||
restartRecovery: true,
|
||||
sameProviderLease: false,
|
||||
}),
|
||||
).toBe("durable_seed");
|
||||
expect(() =>
|
||||
classifyNativeWorkspaceInbound({
|
||||
kind: "existing_run",
|
||||
restartRecovery: false,
|
||||
sameProviderLease: true,
|
||||
}),
|
||||
).toThrow("native_workspace_sync_unexpected_existing_descriptor");
|
||||
});
|
||||
|
||||
it("reads backward-compatible references and the resource disposition", () => {
|
||||
const base = {
|
||||
schema: "paperclip.native-workspace-sync/v1",
|
||||
state: "prepared",
|
||||
descriptorSha256: digest,
|
||||
baselineSha256: digest,
|
||||
finalHostSha256: null,
|
||||
workspaceId: "workspace-1",
|
||||
leaseId: "lease-1",
|
||||
providerLeaseId: "sandbox-1",
|
||||
remoteCwd: "/workspace",
|
||||
};
|
||||
|
||||
expect(readNativeWorkspaceSyncReference(base)).toEqual({
|
||||
...base,
|
||||
resourceDisposition: null,
|
||||
});
|
||||
expect(
|
||||
readNativeWorkspaceSyncReference({
|
||||
...base,
|
||||
resourceDisposition: "keep_running",
|
||||
}),
|
||||
).toEqual({ ...base, resourceDisposition: "keep_running" });
|
||||
expect(
|
||||
readNativeWorkspaceSyncReference({
|
||||
...base,
|
||||
resourceDisposition: "delete_everything",
|
||||
}),
|
||||
).toBeNull();
|
||||
});
|
||||
|
||||
it("rejects traversal before constructing a durable state path", () => {
|
||||
expect(() =>
|
||||
nativeWorkspaceSyncInternals.descriptorPath("../run", digest),
|
||||
).toThrow("native_workspace_sync_invalid_run_id");
|
||||
expect(() =>
|
||||
nativeWorkspaceSyncInternals.descriptorPath("run-1", "../descriptor"),
|
||||
).toThrow("native_workspace_sync_descriptor_digest_invalid");
|
||||
});
|
||||
|
||||
it("writes one immutable descriptor when the same state is replayed", async () => {
|
||||
const paperclipHome = await mkdtemp(
|
||||
path.join(os.tmpdir(), "paperclip-native-workspace-sync-"),
|
||||
);
|
||||
cleanupDirs.push(paperclipHome);
|
||||
process.env.PAPERCLIP_HOME = paperclipHome;
|
||||
process.env.PAPERCLIP_INSTANCE_ID = "descriptor-test";
|
||||
const baseline = {
|
||||
exclude: [".paperclip-runtime"],
|
||||
entries: new Map([
|
||||
[
|
||||
"continuity.txt",
|
||||
{ kind: "file" as const, mode: 0o644, hash: digest },
|
||||
],
|
||||
]),
|
||||
};
|
||||
const descriptor = {
|
||||
schema: "paperclip.native-workspace-sync/v1" as const,
|
||||
binding: {
|
||||
runId: "run-idempotent",
|
||||
companyId: "company-1",
|
||||
workspaceId: "workspace-1",
|
||||
leaseId: "lease-1",
|
||||
providerLeaseId: "sandbox-1",
|
||||
localCwd: path.join(paperclipHome, "workspace"),
|
||||
remoteCwd: "/workspace",
|
||||
},
|
||||
state: "prepared" as const,
|
||||
baselineSha256: directorySnapshotSha256(baseline),
|
||||
baseline: serializeDirectorySnapshot(baseline),
|
||||
gitSnapshot: null,
|
||||
seed: null,
|
||||
createdAt: "2026-01-01T00:00:00.000Z",
|
||||
finalizedAt: null,
|
||||
finalHostSha256: null,
|
||||
resourceDisposition: "keep_running" as const,
|
||||
};
|
||||
|
||||
const first =
|
||||
await nativeWorkspaceSyncInternals.writeDescriptor(descriptor);
|
||||
const second =
|
||||
await nativeWorkspaceSyncInternals.writeDescriptor(descriptor);
|
||||
|
||||
expect(second).toEqual(first);
|
||||
const files = await readdir(
|
||||
path.dirname(
|
||||
nativeWorkspaceSyncInternals.descriptorPath(
|
||||
descriptor.binding.runId,
|
||||
first.descriptorSha256,
|
||||
),
|
||||
),
|
||||
);
|
||||
expect(files.filter((file) => file.endsWith(".json"))).toEqual([
|
||||
`descriptor.${first.descriptorSha256}.json`,
|
||||
]);
|
||||
await expect(
|
||||
nativeWorkspaceSyncInternals.readDescriptor({
|
||||
runId: descriptor.binding.runId,
|
||||
reference: first,
|
||||
}),
|
||||
).resolves.toMatchObject({ descriptor });
|
||||
});
|
||||
|
||||
it("repairs finalized remote and lease stamps after an interrupted commit", async () => {
|
||||
const paperclipHome = await mkdtemp(
|
||||
path.join(os.tmpdir(), "paperclip-native-workspace-sync-repair-"),
|
||||
);
|
||||
cleanupDirs.push(paperclipHome);
|
||||
process.env.PAPERCLIP_HOME = paperclipHome;
|
||||
process.env.PAPERCLIP_INSTANCE_ID = "descriptor-repair-test";
|
||||
const baseline = {
|
||||
exclude: [".paperclip-runtime"],
|
||||
entries: new Map([
|
||||
[
|
||||
"continuity.txt",
|
||||
{ kind: "file" as const, mode: 0o644, hash: digest },
|
||||
],
|
||||
]),
|
||||
};
|
||||
const finalHostSha256 = "b".repeat(64);
|
||||
const descriptor = {
|
||||
schema: "paperclip.native-workspace-sync/v1" as const,
|
||||
binding: {
|
||||
runId: "run-finalized-repair",
|
||||
companyId: "company-1",
|
||||
workspaceId: "workspace-1",
|
||||
leaseId: "lease-1",
|
||||
providerLeaseId: "sandbox-1",
|
||||
localCwd: path.join(paperclipHome, "workspace"),
|
||||
remoteCwd: "/workspace",
|
||||
},
|
||||
state: "finalized" as const,
|
||||
baselineSha256: directorySnapshotSha256(baseline),
|
||||
baseline: serializeDirectorySnapshot(baseline),
|
||||
gitSnapshot: null,
|
||||
seed: null,
|
||||
createdAt: "2026-01-01T00:00:00.000Z",
|
||||
finalizedAt: "2026-01-01T00:01:00.000Z",
|
||||
finalHostSha256,
|
||||
resourceDisposition: "keep_running" as const,
|
||||
};
|
||||
const reference =
|
||||
await nativeWorkspaceSyncInternals.writeDescriptor(descriptor);
|
||||
const rows = (values: unknown[]) => {
|
||||
const query = {
|
||||
from: () => query,
|
||||
where: () => query,
|
||||
for: () => query,
|
||||
limit: () => query,
|
||||
then: <TResult1 = unknown, TResult2 = never>(
|
||||
onfulfilled?:
|
||||
((value: unknown[]) => TResult1 | PromiseLike<TResult1>) | null,
|
||||
onrejected?:
|
||||
((reason: unknown) => TResult2 | PromiseLike<TResult2>) | null,
|
||||
) => Promise.resolve(values).then(onfulfilled, onrejected),
|
||||
};
|
||||
return query;
|
||||
};
|
||||
let persistedLeaseMetadata: Record<string, unknown> | null = null;
|
||||
const db = {
|
||||
select: () =>
|
||||
rows([{ runnerProfileJson: { nativeWorkspaceSync: reference } }]),
|
||||
transaction: async (callback: (tx: unknown) => Promise<unknown>) =>
|
||||
callback({
|
||||
select: () => rows([{ metadata: { retained: true } }]),
|
||||
update: () => ({
|
||||
set: (value: { metadata: Record<string, unknown> }) => ({
|
||||
where: async () => {
|
||||
persistedLeaseMetadata = value.metadata;
|
||||
},
|
||||
}),
|
||||
}),
|
||||
}),
|
||||
};
|
||||
const execute = vi
|
||||
.fn()
|
||||
.mockResolvedValueOnce({ timedOut: false, exitCode: 1, stdout: "" })
|
||||
.mockResolvedValueOnce({ timedOut: false, exitCode: 0, stdout: "" });
|
||||
|
||||
await expect(
|
||||
resumeNativeWorkspaceSync({
|
||||
db: db as never,
|
||||
runId: descriptor.binding.runId,
|
||||
target: {
|
||||
kind: "remote",
|
||||
transport: "sandbox",
|
||||
remoteCwd: descriptor.binding.remoteCwd,
|
||||
sandboxLeaseAcquisition: {
|
||||
providerLeaseId: descriptor.binding.providerLeaseId,
|
||||
},
|
||||
runner: { execute },
|
||||
} as never,
|
||||
}),
|
||||
).resolves.toBe(true);
|
||||
|
||||
expect(execute).toHaveBeenCalledTimes(2);
|
||||
expect(persistedLeaseMetadata).toMatchObject({
|
||||
retained: true,
|
||||
nativeWorkspaceSync: {
|
||||
schema: "paperclip.native-workspace-stamp/v1",
|
||||
workspaceId: descriptor.binding.workspaceId,
|
||||
providerLeaseId: descriptor.binding.providerLeaseId,
|
||||
remoteCwd: descriptor.binding.remoteCwd,
|
||||
hostSha256: finalHostSha256,
|
||||
finalizedRunId: descriptor.binding.runId,
|
||||
},
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -2060,6 +2060,11 @@ function createSandboxEnvironmentDriver(
|
||||
...(reusableLease?.metadata?.nativeHarnessBackup
|
||||
? { nativeHarnessBackup: reusableLease.metadata.nativeHarnessBackup }
|
||||
: {}),
|
||||
...(providerLease && reusableLease?.metadata?.nativeWorkspaceSync
|
||||
? {
|
||||
nativeWorkspaceSync: reusableLease.metadata.nativeWorkspaceSync,
|
||||
}
|
||||
: {}),
|
||||
...(reusableScope ? { reusableSandboxLease: reusableScope } : {}),
|
||||
};
|
||||
try {
|
||||
@@ -2282,6 +2287,13 @@ function createSandboxEnvironmentDriver(
|
||||
...(reusableLease?.metadata?.nativeHarnessBackup
|
||||
? { nativeHarnessBackup: reusableLease.metadata.nativeHarnessBackup }
|
||||
: {}),
|
||||
...(reusableLease &&
|
||||
providerLease.providerLeaseId === reusableLease.providerLeaseId &&
|
||||
reusableLease.metadata?.nativeWorkspaceSync
|
||||
? {
|
||||
nativeWorkspaceSync: reusableLease.metadata.nativeWorkspaceSync,
|
||||
}
|
||||
: {}),
|
||||
...(reusableScope ? { reusableSandboxLease: reusableScope } : {}),
|
||||
};
|
||||
try {
|
||||
@@ -3154,6 +3166,7 @@ const INTERNAL_PLUGIN_SANDBOX_CONFIG_KEYS = new Set([
|
||||
"sandboxProviderPlugin",
|
||||
"sandboxLeaseAcquisition",
|
||||
"nativeHarnessBackup",
|
||||
"nativeWorkspaceSync",
|
||||
]);
|
||||
|
||||
// Drop the host-internal and per-lease runtime keys from a sandbox config
|
||||
|
||||
@@ -142,6 +142,9 @@ import {
|
||||
materializeLegacyQuestionResponseWakeProjection,
|
||||
materializeNativeInteractionResponses,
|
||||
NativeCancellationPendingRecoveryError,
|
||||
prepareNativeWorkspaceSync,
|
||||
readNativeWorkspaceSyncReference,
|
||||
recordNativeFinalizationFailure,
|
||||
type NativeRestartRecoveryClaim,
|
||||
rebindNativeSessionCheckpoint,
|
||||
reconcileNativeFinalizations,
|
||||
@@ -1746,12 +1749,53 @@ export function leaseReleaseStatusForRunStatus(
|
||||
return status === "failed" || status === "timed_out" ? "failed" : "released";
|
||||
}
|
||||
|
||||
export function providerResourceDispositionForTerminalRun(
|
||||
desired: ProviderResourceDisposition | undefined,
|
||||
status: string | null | undefined,
|
||||
): ProviderResourceDisposition | undefined {
|
||||
return desired === "keep_running" && status !== "succeeded"
|
||||
? "stop_and_retain"
|
||||
: desired;
|
||||
}
|
||||
|
||||
export interface NativeSandboxLifecycle {
|
||||
runnerProcess: "per_turn" | "warm";
|
||||
sandboxResource: "keep_running" | "stop_and_reuse" | "destroy_after_turn";
|
||||
failoverBackup: "verified";
|
||||
}
|
||||
|
||||
export function resolveReusableSandboxLifecycle(input: {
|
||||
lifecyclePolicy:
|
||||
| { mode: "per_turn"; idleTimeoutMs: null }
|
||||
| { mode: "warm"; idleTimeoutMs: number };
|
||||
target: {
|
||||
kind: "local" | "remote";
|
||||
transport?: string;
|
||||
reusableLeaseConfigured?: boolean;
|
||||
effectiveCapabilities?: { reusableLeases: boolean } | null;
|
||||
} | null;
|
||||
}): NativeSandboxLifecycle | null {
|
||||
if (input.target?.kind !== "remote" || input.target.transport !== "sandbox") {
|
||||
return null;
|
||||
}
|
||||
const reusableLease =
|
||||
input.target.reusableLeaseConfigured === true &&
|
||||
input.target.effectiveCapabilities?.reusableLeases === true;
|
||||
if (input.lifecyclePolicy.mode === "warm" && !reusableLease) {
|
||||
throw new Error("runner_warm_lifecycle_requires_reusable_provider_lease");
|
||||
}
|
||||
return {
|
||||
runnerProcess: input.lifecyclePolicy.mode,
|
||||
sandboxResource:
|
||||
input.lifecyclePolicy.mode === "warm"
|
||||
? "keep_running"
|
||||
: reusableLease
|
||||
? "stop_and_reuse"
|
||||
: "destroy_after_turn",
|
||||
failoverBackup: "verified",
|
||||
};
|
||||
}
|
||||
|
||||
export function resolveNativeSandboxLifecycle(input: {
|
||||
adapterType: string;
|
||||
lifecyclePolicy:
|
||||
@@ -1770,22 +1814,7 @@ export function resolveNativeSandboxLifecycle(input: {
|
||||
input.target.transport !== "sandbox"
|
||||
)
|
||||
return null;
|
||||
const reusableLease =
|
||||
input.target.reusableLeaseConfigured === true &&
|
||||
input.target.effectiveCapabilities?.reusableLeases === true;
|
||||
if (input.lifecyclePolicy.mode === "warm" && !reusableLease) {
|
||||
throw new Error("runner_warm_lifecycle_requires_reusable_provider_lease");
|
||||
}
|
||||
return {
|
||||
runnerProcess: input.lifecyclePolicy.mode,
|
||||
sandboxResource:
|
||||
input.lifecyclePolicy.mode === "warm"
|
||||
? "keep_running"
|
||||
: reusableLease
|
||||
? "stop_and_reuse"
|
||||
: "destroy_after_turn",
|
||||
failoverBackup: "verified",
|
||||
};
|
||||
return resolveReusableSandboxLifecycle(input);
|
||||
}
|
||||
|
||||
export function applyPersistedExecutionWorkspaceConfig(input: {
|
||||
@@ -8170,6 +8199,18 @@ class NativeSessionResumeScheduledError extends Error {
|
||||
}
|
||||
}
|
||||
|
||||
class NativeWorkspaceFinalizeScheduledError extends Error {
|
||||
constructor(
|
||||
readonly original: unknown,
|
||||
readonly terminalFailure: boolean,
|
||||
readonly reasonCode:
|
||||
"workspace_sync_out_failed" | "workspace_sync_out_unrecoverable",
|
||||
) {
|
||||
super("Native workspace finalization recovery has been scheduled.");
|
||||
this.name = "NativeWorkspaceFinalizeScheduledError";
|
||||
}
|
||||
}
|
||||
|
||||
type WorkspaceReadyCommentWriter = {
|
||||
addComment: (
|
||||
issueId: string,
|
||||
@@ -12945,7 +12986,10 @@ export function heartbeatService(
|
||||
// A result committed before the old controller stopped outranks process
|
||||
// recovery. Finish its durable workspace/status suffix before deciding
|
||||
// whether any provider authority needs to be reopened.
|
||||
await reconcileNativeFinalizations(db);
|
||||
await reconcileNativeFinalizations(db, undefined, {
|
||||
environmentRuntime,
|
||||
onWorkspaceSettled: settleRecoveredNativeWorkspace,
|
||||
});
|
||||
const intent = await readHotRestartIntent().catch((error) => {
|
||||
logger.warn(
|
||||
{ err: error },
|
||||
@@ -16854,6 +16898,41 @@ export function heartbeatService(
|
||||
return blocked;
|
||||
}
|
||||
|
||||
async function settleRecoveredNativeWorkspace(input: {
|
||||
runId: string;
|
||||
companyId: string;
|
||||
agentId: string;
|
||||
succeeded: boolean;
|
||||
}) {
|
||||
const settledRun = await getRun(input.runId);
|
||||
const workspaceSyncReference = readNativeWorkspaceSyncReference(
|
||||
parseObject(settledRun?.runnerProfileJson).nativeWorkspaceSync,
|
||||
);
|
||||
await releaseEnvironmentLeasesForRun({
|
||||
runId: input.runId,
|
||||
companyId: input.companyId,
|
||||
agentId: input.agentId,
|
||||
status: settledRun?.status,
|
||||
failureReason: settledRun?.error ?? undefined,
|
||||
providerResourceDisposition: input.succeeded
|
||||
? (workspaceSyncReference?.resourceDisposition ?? "stop_and_retain")
|
||||
: "stop_and_retain",
|
||||
});
|
||||
await releaseRuntimeServicesForRun(input.runId).catch(() => undefined);
|
||||
await finalizeAgentStatus(
|
||||
input.agentId,
|
||||
input.succeeded ? "succeeded" : "failed",
|
||||
input.succeeded
|
||||
? null
|
||||
: (settledRun?.error ?? "native_workspace_sync_out_failed"),
|
||||
{
|
||||
wasFirstHeartbeat: settledRun
|
||||
? timerClaimWasFirstHeartbeat(settledRun)
|
||||
: undefined,
|
||||
},
|
||||
).catch(() => undefined);
|
||||
}
|
||||
|
||||
async function reapOrphanedRuns(opts?: { staleThresholdMs?: number }) {
|
||||
const staleThresholdMs = opts?.staleThresholdMs ?? 0;
|
||||
const now = new Date();
|
||||
@@ -16861,7 +16940,10 @@ export function heartbeatService(
|
||||
// Complete persisted native results before generic orphan recovery. The
|
||||
// reconciler reads the durable workspace barrier and persisted runtime
|
||||
// mode, never the current feature flag.
|
||||
await reconcileNativeFinalizations(db).catch((error) => {
|
||||
await reconcileNativeFinalizations(db, undefined, {
|
||||
environmentRuntime,
|
||||
onWorkspaceSettled: settleRecoveredNativeWorkspace,
|
||||
}).catch((error) => {
|
||||
logger.warn(
|
||||
{ err: error },
|
||||
"failed to reconcile persisted native finalizations before orphan reaping",
|
||||
@@ -17876,6 +17958,10 @@ export function heartbeatService(
|
||||
activeRunExecutions.add(run.id);
|
||||
let runScratch: HeartbeatRunScratch | null = null;
|
||||
let nativeSessionResumeScheduled = false;
|
||||
let nativeWorkspaceFinalizeScheduled = false;
|
||||
let nativeWorkspaceSync: Awaited<
|
||||
ReturnType<typeof prepareNativeWorkspaceSync>
|
||||
> = null;
|
||||
let providerResourceDispositionForRun:
|
||||
ProviderResourceDisposition | undefined;
|
||||
let nativeLifecycleTelemetryForRun:
|
||||
@@ -19659,6 +19745,9 @@ export function heartbeatService(
|
||||
driver: selectedEnvironment.driver,
|
||||
leaseId: activeEnvironmentLease.lease.id,
|
||||
workspaceRealization,
|
||||
sandboxLeaseAcquisition:
|
||||
activeEnvironmentLease.lease.metadata?.sandboxLeaseAcquisition ??
|
||||
null,
|
||||
...(typeof activeEnvironmentLease.lease.metadata?.remoteCwd === "string"
|
||||
? {
|
||||
remoteCwd: activeEnvironmentLease.lease.metadata.remoteCwd,
|
||||
@@ -20363,6 +20452,9 @@ export function heartbeatService(
|
||||
agentId: heartbeatRuns.agentId,
|
||||
runnerInstanceId: heartbeatRuns.runnerInstanceId,
|
||||
nativeSessionId: heartbeatRuns.nativeSessionId,
|
||||
processPid: heartbeatRuns.processPid,
|
||||
processGroupId: heartbeatRuns.processGroupId,
|
||||
processStartedAt: heartbeatRuns.processStartedAt,
|
||||
runnerProfileJson: heartbeatRuns.runnerProfileJson,
|
||||
})
|
||||
.from(heartbeatRuns)
|
||||
@@ -20718,6 +20810,21 @@ export function heartbeatService(
|
||||
? previousNativeRun.runnerInstanceId
|
||||
: (lockedRun.runnerInstanceId ?? nativeRunnerInstanceId),
|
||||
nativeSessionId: lockedRun.nativeSessionId ?? nativeSessionId,
|
||||
processPid:
|
||||
lockedRun.processPid ??
|
||||
(previousNativeRun?.nativeSessionId === nativeSessionId
|
||||
? previousNativeRun.processPid
|
||||
: null),
|
||||
processGroupId:
|
||||
lockedRun.processGroupId ??
|
||||
(previousNativeRun?.nativeSessionId === nativeSessionId
|
||||
? previousNativeRun.processGroupId
|
||||
: null),
|
||||
processStartedAt:
|
||||
lockedRun.processStartedAt ??
|
||||
(previousNativeRun?.nativeSessionId === nativeSessionId
|
||||
? previousNativeRun.processStartedAt
|
||||
: null),
|
||||
nativeIssueId: lockedRun.nativeIssueId ?? issueRef.id,
|
||||
driverKind:
|
||||
lockedRun.driverKind ??
|
||||
@@ -20745,7 +20852,30 @@ export function heartbeatService(
|
||||
})
|
||||
.onConflictDoNothing();
|
||||
});
|
||||
nativeWorkspaceSync = await prepareNativeWorkspaceSync({
|
||||
db,
|
||||
runId: run.id,
|
||||
companyId: agent.companyId,
|
||||
workspaceId: nativeExecutionWorkspaceId,
|
||||
workspaceLocalDir: executionWorkspace.cwd,
|
||||
target: executionTarget,
|
||||
lease: activeEnvironmentLease.lease,
|
||||
restartRecovery: runOptions.nativeRestartRecovery,
|
||||
resourceDisposition: providerResourceDispositionForRun,
|
||||
});
|
||||
} else {
|
||||
const legacyWarmLifecycle =
|
||||
executionTarget?.kind === "remote" &&
|
||||
executionTarget.transport === "sandbox" &&
|
||||
executionTarget.runnerLifecyclePolicy?.mode === "warm"
|
||||
? resolveReusableSandboxLifecycle({
|
||||
lifecyclePolicy: executionTarget.runnerLifecyclePolicy,
|
||||
target: executionTarget,
|
||||
})
|
||||
: null;
|
||||
if (legacyWarmLifecycle?.sandboxResource === "keep_running") {
|
||||
providerResourceDispositionForRun = "keep_running";
|
||||
}
|
||||
await db
|
||||
.update(heartbeatRuns)
|
||||
.set({
|
||||
@@ -21315,19 +21445,26 @@ export function heartbeatService(
|
||||
// If recording the barrier itself fails, propagate as a run failure
|
||||
// rather than silently leaving dependents stranded behind a missing
|
||||
// finalize row.
|
||||
if (nativeWorkspaceSync) {
|
||||
await nativeWorkspaceSync.restoreWorkspace();
|
||||
}
|
||||
await recordWorkspaceFinalize("succeeded");
|
||||
if (adapterResult.nativeFinalization) {
|
||||
adapterResult.nativeFinalization.workspaceFinalizeStatus =
|
||||
"succeeded";
|
||||
try {
|
||||
await finalizeNativeRun({
|
||||
const finalized = await finalizeNativeRun({
|
||||
db,
|
||||
runId: run.id,
|
||||
workspaceFinalizeStatus: "succeeded",
|
||||
preserveProviderAttempt: Boolean(nativeWorkspaceSync),
|
||||
});
|
||||
await dispatchPendingNativeStatusWakeups({
|
||||
companyId: run.companyId,
|
||||
});
|
||||
if (finalized.phase === "committed") {
|
||||
await nativeWorkspaceSync?.cleanup();
|
||||
}
|
||||
} catch (finalizeErr) {
|
||||
logger.warn(
|
||||
{ err: finalizeErr, runId: run.id },
|
||||
@@ -21379,6 +21516,40 @@ export function heartbeatService(
|
||||
);
|
||||
}
|
||||
if (nativeRuntimeResolution.kind === "native") {
|
||||
const proposedResult = await db
|
||||
.select({ resultId: nativeRunFinalizations.resultId })
|
||||
.from(nativeRunFinalizations)
|
||||
.where(eq(nativeRunFinalizations.runId, run.id))
|
||||
.limit(1)
|
||||
.then((rows) => rows[0]?.resultId ?? null);
|
||||
if (proposedResult && nativeWorkspaceSync) {
|
||||
const workspaceFailureMessage =
|
||||
adapterErr instanceof Error ? adapterErr.message : "";
|
||||
const unrecoverable =
|
||||
workspaceFailureMessage ===
|
||||
"workspace_sync_out_unrecoverable" ||
|
||||
workspaceFailureMessage.includes("daytona_sandbox_not_found");
|
||||
const failure = await recordNativeFinalizationFailure({
|
||||
db,
|
||||
runId: run.id,
|
||||
error: new Error(
|
||||
unrecoverable
|
||||
? "native_workspace_sync_out_unrecoverable"
|
||||
: "native_workspace_sync_out_failed",
|
||||
),
|
||||
projectRunStatus: true,
|
||||
failureScope: "workspace",
|
||||
permanent: unrecoverable,
|
||||
});
|
||||
nativeWorkspaceFinalizeScheduled = true;
|
||||
throw new NativeWorkspaceFinalizeScheduledError(
|
||||
adapterErr,
|
||||
failure.phase === "terminal_failure",
|
||||
unrecoverable
|
||||
? "workspace_sync_out_unrecoverable"
|
||||
: "workspace_sync_out_failed",
|
||||
);
|
||||
}
|
||||
try {
|
||||
await finalizeNativeRun({
|
||||
db,
|
||||
@@ -22064,6 +22235,45 @@ export function heartbeatService(
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (err instanceof NativeWorkspaceFinalizeScheduledError) {
|
||||
const coordinator = await db
|
||||
.select({
|
||||
nextAttemptAt: nativeRunFinalizations.nextAttemptAt,
|
||||
attempt: nativeRunFinalizations.attempt,
|
||||
})
|
||||
.from(nativeRunFinalizations)
|
||||
.where(eq(nativeRunFinalizations.runId, run.id))
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
await appendRunEvent(run, {
|
||||
eventType: "lifecycle",
|
||||
stream: "system",
|
||||
level: err.terminalFailure ? "error" : "warn",
|
||||
message: err.terminalFailure
|
||||
? "native result is durable, but the sandbox containing unexported workspace changes is unrecoverable"
|
||||
: "native result is durable; workspace copy-back will retry without another provider turn",
|
||||
payload: {
|
||||
attempt: coordinator?.attempt ?? null,
|
||||
nextAttemptAt: coordinator?.nextAttemptAt?.toISOString() ?? null,
|
||||
fallbackSuppressed: true,
|
||||
retryReasonCode: err.reasonCode,
|
||||
},
|
||||
}).catch(() => undefined);
|
||||
if (err.terminalFailure) {
|
||||
// The durable coordinator already failed the run, blocked the
|
||||
// issue, and cleared its execution lock. Let ordinary teardown
|
||||
// release the now-useless lease and return the agent to service.
|
||||
nativeWorkspaceFinalizeScheduled = false;
|
||||
providerResourceDispositionForRun = "stop_and_retain";
|
||||
await finalizeAgentStatus(
|
||||
run.agentId,
|
||||
"failed",
|
||||
"native_workspace_sync_out_unrecoverable",
|
||||
{ wasFirstHeartbeat: timerClaimWasFirstHeartbeat(run) },
|
||||
).catch(() => undefined);
|
||||
}
|
||||
return;
|
||||
}
|
||||
const message = redactCurrentUserText(
|
||||
err instanceof Error ? err.message : "Unknown adapter failure",
|
||||
await getCurrentUserRedactionOptions(),
|
||||
@@ -22456,7 +22666,11 @@ export function heartbeatService(
|
||||
// terminal". When the teardown reaches this point with the run still
|
||||
// running or queued, force a terminal status before the lease is
|
||||
// released, so the UI never shows a finished task as "Live".
|
||||
if (latestRun && !nativeSessionResumeScheduled) {
|
||||
if (
|
||||
latestRun &&
|
||||
!nativeSessionResumeScheduled &&
|
||||
!nativeWorkspaceFinalizeScheduled
|
||||
) {
|
||||
latestRun = await terminalizeRunOnLeaseRelease(latestRun).catch(
|
||||
(terminalizeErr) => {
|
||||
logger.error(
|
||||
@@ -22467,7 +22681,15 @@ export function heartbeatService(
|
||||
},
|
||||
);
|
||||
}
|
||||
if (!nativeSessionResumeScheduled) {
|
||||
// Warm retention is earned only by a fully successful turn. A failed,
|
||||
// cancelled, or timed-out run stops the reusable sandbox so the next
|
||||
// acquisition must revalidate and explicitly resume it.
|
||||
providerResourceDispositionForRun =
|
||||
providerResourceDispositionForTerminalRun(
|
||||
providerResourceDispositionForRun,
|
||||
latestRun?.status,
|
||||
);
|
||||
if (!nativeSessionResumeScheduled && !nativeWorkspaceFinalizeScheduled) {
|
||||
await releaseEnvironmentLeasesForRun({
|
||||
runId: run.id,
|
||||
companyId: run.companyId,
|
||||
@@ -22539,7 +22761,11 @@ export function heartbeatService(
|
||||
}
|
||||
}
|
||||
activeRunExecutions.delete(run.id);
|
||||
if (!nativeSessionResumeScheduled && !shutdownInProgress) {
|
||||
if (
|
||||
!nativeSessionResumeScheduled &&
|
||||
!nativeWorkspaceFinalizeScheduled &&
|
||||
!shutdownInProgress
|
||||
) {
|
||||
await startNextQueuedRunForAgent(run.agentId);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -9,4 +9,5 @@ export * from "./paperclip-control-plane-port.js";
|
||||
export * from "./native-run-finalizer.js";
|
||||
export * from "./native-finalization-reconciler.js";
|
||||
export * from "./native-restart-recovery.js";
|
||||
export * from "./native-workspace-sync.js";
|
||||
export * from "./status-arbiter.js";
|
||||
@@ -24,6 +24,11 @@ import { issueRecoveryActionService } from "../issue-recovery-actions.js";
|
||||
import { issueService } from "../issues.js";
|
||||
import { emitAgentTaskRun } from "../agent-task-run-telemetry.js";
|
||||
import { resumeNativeWorkspaceFinalization } from "./native-workspace-finalizer.js";
|
||||
import {
|
||||
cleanupNativeWorkspaceSync,
|
||||
readNativeWorkspaceSyncReference,
|
||||
} from "./native-workspace-sync.js";
|
||||
import type { EnvironmentRuntimeService } from "../environment-runtime.js";
|
||||
import { classifyNativeEvidence } from "./evidence-classifier.js";
|
||||
import { recordNativeWorkAssessment } from "./work-assessments.js";
|
||||
import {
|
||||
@@ -373,19 +378,33 @@ export async function claimNativeSessionResumptions(input: {
|
||||
}
|
||||
|
||||
/** Recovery is keyed only by persisted mode/coordinator state, never the live flag. */
|
||||
export async function reconcileNativeFinalizations(db: Db, runIds?: string[]) {
|
||||
const rows = await db.select({
|
||||
runId: heartbeatRuns.id,
|
||||
companyId: heartbeatRuns.companyId,
|
||||
agentId: heartbeatRuns.agentId,
|
||||
issueId: nativeRunFinalizations.issueId,
|
||||
issueStatus: issues.status,
|
||||
issueStatusVersion: issues.statusVersion,
|
||||
issueDecisionId: issues.lastStatusDecisionId,
|
||||
coordinatorPhase: nativeRunFinalizations.phase,
|
||||
assessmentId: nativeRunFinalizations.assessmentId,
|
||||
decisionId: nativeRunFinalizations.decisionId,
|
||||
})
|
||||
export async function reconcileNativeFinalizations(
|
||||
db: Db,
|
||||
runIds?: string[],
|
||||
options: {
|
||||
environmentRuntime?: EnvironmentRuntimeService;
|
||||
onWorkspaceSettled?: (input: {
|
||||
runId: string;
|
||||
companyId: string;
|
||||
agentId: string;
|
||||
succeeded: boolean;
|
||||
}) => Promise<void>;
|
||||
} = {},
|
||||
) {
|
||||
const rows = await db
|
||||
.select({
|
||||
runId: heartbeatRuns.id,
|
||||
companyId: heartbeatRuns.companyId,
|
||||
agentId: heartbeatRuns.agentId,
|
||||
issueId: nativeRunFinalizations.issueId,
|
||||
issueStatus: issues.status,
|
||||
issueStatusVersion: issues.statusVersion,
|
||||
issueDecisionId: issues.lastStatusDecisionId,
|
||||
coordinatorPhase: nativeRunFinalizations.phase,
|
||||
assessmentId: nativeRunFinalizations.assessmentId,
|
||||
decisionId: nativeRunFinalizations.decisionId,
|
||||
runnerProfileJson: heartbeatRuns.runnerProfileJson,
|
||||
})
|
||||
.from(heartbeatRuns)
|
||||
.innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id))
|
||||
.innerJoin(issues, and(
|
||||
@@ -617,14 +636,59 @@ export async function reconcileNativeFinalizations(db: Db, runIds?: string[]) {
|
||||
if (!recoveryDecision.effects.some((effect) => effect.kind === "resume_workspace_operation")) {
|
||||
throw new Error("native_reconciliation_workspace_resume_policy_missing");
|
||||
}
|
||||
const operation = await resumeNativeWorkspaceFinalization({ db, runId: row.runId });
|
||||
const workspaceFinalizeStatus = operation.status === "succeeded" ? "succeeded" : "failed";
|
||||
const operation = await resumeNativeWorkspaceFinalization({
|
||||
db,
|
||||
runId: row.runId,
|
||||
environmentRuntime: options.environmentRuntime,
|
||||
});
|
||||
const workspaceFinalizeStatus =
|
||||
operation.status === "succeeded" ? "succeeded" : "failed";
|
||||
if (workspaceFinalizeStatus === "failed") {
|
||||
const unrecoverable = operation.stderrExcerpt?.includes(
|
||||
"workspace_sync_out_unrecoverable",
|
||||
);
|
||||
const failure = await recordNativeFinalizationFailure({
|
||||
db,
|
||||
runId: row.runId,
|
||||
error: new Error(
|
||||
unrecoverable
|
||||
? "native_workspace_sync_out_unrecoverable"
|
||||
: "native_workspace_sync_out_failed",
|
||||
),
|
||||
projectRunStatus: true,
|
||||
failureScope: "workspace",
|
||||
permanent: unrecoverable,
|
||||
});
|
||||
results.push({
|
||||
...failure,
|
||||
reconciliationAction: "resume_workspace_operation" as const,
|
||||
workspaceOperationId: operation.id,
|
||||
workspaceFinalizeStatus,
|
||||
reconciliationDecision: recoveryDecision,
|
||||
});
|
||||
if (failure.phase === "terminal_failure") {
|
||||
await options
|
||||
.onWorkspaceSettled?.({
|
||||
runId: row.runId,
|
||||
companyId: row.companyId,
|
||||
agentId: row.agentId,
|
||||
succeeded: false,
|
||||
})
|
||||
.catch(() => undefined);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
const finalized = await finalizeNativeRun({
|
||||
db,
|
||||
runId: row.runId,
|
||||
workspaceFinalizeStatus,
|
||||
projectRunStatus: true,
|
||||
preserveProviderAttempt: Boolean(
|
||||
readNativeWorkspaceSyncReference(
|
||||
record(row.runnerProfileJson).nativeWorkspaceSync,
|
||||
),
|
||||
),
|
||||
});
|
||||
results.push({
|
||||
...finalized,
|
||||
@@ -633,6 +697,20 @@ export async function reconcileNativeFinalizations(db: Db, runIds?: string[]) {
|
||||
workspaceFinalizeStatus,
|
||||
reconciliationDecision: recoveryDecision,
|
||||
});
|
||||
if (
|
||||
workspaceFinalizeStatus === "succeeded" &&
|
||||
finalized.phase === "committed"
|
||||
) {
|
||||
await cleanupNativeWorkspaceSync(row.runId).catch(() => undefined);
|
||||
await options
|
||||
.onWorkspaceSettled?.({
|
||||
runId: row.runId,
|
||||
companyId: row.companyId,
|
||||
agentId: row.agentId,
|
||||
succeeded: true,
|
||||
})
|
||||
.catch(() => undefined);
|
||||
}
|
||||
} catch (error) {
|
||||
const failure = await recordNativeFinalizationFailure({
|
||||
db,
|
||||
@@ -659,12 +737,29 @@ export async function reconcileNativeFinalizations(db: Db, runIds?: string[]) {
|
||||
throw new Error("native_reconciliation_replay_policy_invalid");
|
||||
}
|
||||
try {
|
||||
results.push(await finalizeNativeRun({
|
||||
const finalized = await finalizeNativeRun({
|
||||
db,
|
||||
runId: row.runId,
|
||||
workspaceFinalizeStatus: barrier.status as "succeeded" | "failed",
|
||||
projectRunStatus: true,
|
||||
}));
|
||||
preserveProviderAttempt: Boolean(
|
||||
readNativeWorkspaceSyncReference(
|
||||
record(row.runnerProfileJson).nativeWorkspaceSync,
|
||||
),
|
||||
),
|
||||
});
|
||||
results.push(finalized);
|
||||
if (finalized.phase === "committed") {
|
||||
await cleanupNativeWorkspaceSync(row.runId).catch(() => undefined);
|
||||
await options
|
||||
.onWorkspaceSettled?.({
|
||||
runId: row.runId,
|
||||
companyId: row.companyId,
|
||||
agentId: row.agentId,
|
||||
succeeded: true,
|
||||
})
|
||||
.catch(() => undefined);
|
||||
}
|
||||
} catch (error) {
|
||||
results.push(await recordNativeFinalizationFailure({
|
||||
db,
|
||||
|
||||
@@ -279,6 +279,93 @@ describeEmbeddedPostgres("native run finalizer / status decision committer — a
|
||||
expect(run?.status).toBe("running");
|
||||
});
|
||||
|
||||
it("bounds workspace-only retries without consuming the provider attempt", async () => {
|
||||
const fixture = await seedNativeRun();
|
||||
await db.insert(nativeRunFinalizations).values({
|
||||
runId: fixture.runId,
|
||||
companyId,
|
||||
issueId: fixture.issueId,
|
||||
phase: "observed",
|
||||
attempt: 1,
|
||||
});
|
||||
|
||||
for (
|
||||
let expectedAttempt = 1;
|
||||
expectedAttempt <= 3;
|
||||
expectedAttempt += 1
|
||||
) {
|
||||
await recordNativeFinalizationFailure({
|
||||
db,
|
||||
runId: fixture.runId,
|
||||
error: new Error("native_workspace_sync_out_failed"),
|
||||
projectRunStatus: true,
|
||||
failureScope: "workspace",
|
||||
});
|
||||
const coordinator = await db
|
||||
.select()
|
||||
.from(nativeRunFinalizations)
|
||||
.where(eq(nativeRunFinalizations.runId, fixture.runId))
|
||||
.then((rows) => rows[0]!);
|
||||
expect(coordinator.attempt).toBe(1);
|
||||
expect(coordinator.failureDetail).toMatchObject({
|
||||
workspaceFinalizeAttempt: expectedAttempt,
|
||||
});
|
||||
expect(coordinator.phase).toBe(
|
||||
expectedAttempt === 3 ? "terminal_failure" : "retryable_failure",
|
||||
);
|
||||
}
|
||||
|
||||
await expect(
|
||||
db
|
||||
.select({ status: heartbeatRuns.status })
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.id, fixture.runId))
|
||||
.then((rows) => rows[0]?.status),
|
||||
).resolves.toBe("failed");
|
||||
});
|
||||
|
||||
it("blocks immediately when the sandbox with unexported changes is gone", async () => {
|
||||
const fixture = await seedNativeRun();
|
||||
await db.insert(nativeRunFinalizations).values({
|
||||
runId: fixture.runId,
|
||||
companyId,
|
||||
issueId: fixture.issueId,
|
||||
phase: "workspace_finalizing",
|
||||
attempt: 1,
|
||||
resultId: null,
|
||||
});
|
||||
|
||||
const failure = await recordNativeFinalizationFailure({
|
||||
db,
|
||||
runId: fixture.runId,
|
||||
error: new Error("native_workspace_sync_out_unrecoverable"),
|
||||
projectRunStatus: true,
|
||||
failureScope: "workspace",
|
||||
permanent: true,
|
||||
});
|
||||
|
||||
expect(failure).toMatchObject({
|
||||
phase: "terminal_failure",
|
||||
failureCode: "native_workspace_sync_out_unrecoverable",
|
||||
nextAttemptAt: null,
|
||||
attempt: 1,
|
||||
});
|
||||
await expect(
|
||||
db
|
||||
.select({ status: heartbeatRuns.status })
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.id, fixture.runId))
|
||||
.then((rows) => rows[0]?.status),
|
||||
).resolves.toBe("failed");
|
||||
await expect(
|
||||
db
|
||||
.select({ status: issues.status })
|
||||
.from(issues)
|
||||
.where(eq(issues.id, fixture.issueId))
|
||||
.then((rows) => rows[0]?.status),
|
||||
).resolves.toBe("blocked");
|
||||
});
|
||||
|
||||
it("emits exactly one event for a cancel_continuations write (trap 2: :685/:518 overlap)", async () => {
|
||||
const fixture = await seedNativeRun();
|
||||
// Build the minimal real rows commitNativeStatusDecision's foreign keys
|
||||
@@ -329,7 +416,10 @@ describeEmbeddedPostgres("native run finalizer / status decision committer — a
|
||||
toStatus: "cancelled",
|
||||
reasonCode: "cancellation_issue_authorized",
|
||||
unblockDescriptor: null,
|
||||
effects: [{ kind: "release_checkout" }, { kind: "cancel_continuations" }],
|
||||
effects: [
|
||||
{ kind: "release_checkout" },
|
||||
{ kind: "cancel_continuations" },
|
||||
],
|
||||
};
|
||||
|
||||
const callsBefore = mockTelemetryClient.track.mock.calls.length;
|
||||
|
||||
@@ -126,7 +126,11 @@ async function acceptedInteractionFromRun(input: {
|
||||
.then((rows) => rows[0] ?? null);
|
||||
}
|
||||
|
||||
async function claimCoordinator(input: { db: Db; runId: string }) {
|
||||
async function claimCoordinator(input: {
|
||||
db: Db;
|
||||
runId: string;
|
||||
preserveProviderAttempt?: boolean;
|
||||
}) {
|
||||
const leaseOwner = `native-finalizer:${randomUUID()}`;
|
||||
const now = new Date();
|
||||
const claimed = await input.db.transaction(async (tx) => {
|
||||
@@ -152,23 +156,33 @@ async function claimCoordinator(input: { db: Db; runId: string }) {
|
||||
}
|
||||
if (coordinator.phase === "terminal_failure") throw new Error("native_finalization_terminal_failure");
|
||||
if (
|
||||
coordinator.leaseOwner
|
||||
&& coordinator.leaseExpiresAt
|
||||
&& coordinator.leaseExpiresAt > now
|
||||
&& coordinator.leaseOwner !== leaseOwner
|
||||
) throw new Error("native_finalization_lease_busy");
|
||||
const [updated] = await tx.update(nativeRunFinalizations).set({
|
||||
leaseOwner,
|
||||
leaseExpiresAt: new Date(now.getTime() + 5 * 60_000),
|
||||
attempt: coordinator.attempt + 1,
|
||||
phase: coordinator.phase === "retryable_failure"
|
||||
? coordinator.assessmentId ? "arbitrating" : "workspace_finalizing"
|
||||
: coordinator.phase,
|
||||
failureCode: null,
|
||||
failureDetail: null,
|
||||
nextAttemptAt: null,
|
||||
updatedAt: now,
|
||||
}).where(eq(nativeRunFinalizations.runId, input.runId)).returning();
|
||||
coordinator.leaseOwner &&
|
||||
coordinator.leaseExpiresAt &&
|
||||
coordinator.leaseExpiresAt > now &&
|
||||
coordinator.leaseOwner !== leaseOwner
|
||||
)
|
||||
throw new Error("native_finalization_lease_busy");
|
||||
const [updated] = await tx
|
||||
.update(nativeRunFinalizations)
|
||||
.set({
|
||||
leaseOwner,
|
||||
leaseExpiresAt: new Date(now.getTime() + 5 * 60_000),
|
||||
attempt: input.preserveProviderAttempt
|
||||
? coordinator.attempt
|
||||
: coordinator.attempt + 1,
|
||||
phase:
|
||||
coordinator.phase === "retryable_failure"
|
||||
? coordinator.assessmentId
|
||||
? "arbitrating"
|
||||
: "workspace_finalizing"
|
||||
: coordinator.phase,
|
||||
failureCode: null,
|
||||
failureDetail: null,
|
||||
nextAttemptAt: null,
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(eq(nativeRunFinalizations.runId, input.runId))
|
||||
.returning();
|
||||
if (!updated) throw new Error("native_finalization_claim_failed");
|
||||
return { coordinator: updated, leaseOwner };
|
||||
});
|
||||
@@ -183,6 +197,8 @@ async function recordRetryableFailure(input: {
|
||||
message: string;
|
||||
nextAction: string;
|
||||
projectRunStatus?: boolean;
|
||||
failureScope?: "provider" | "workspace";
|
||||
permanent?: boolean;
|
||||
}) {
|
||||
const now = new Date();
|
||||
const nextAttemptAt = new Date(now.getTime() + 30_000);
|
||||
@@ -228,50 +244,87 @@ async function recordRetryableFailure(input: {
|
||||
const supersededByNewerRun = Boolean(
|
||||
latestDecisionRun && latestDecisionRun.createdAt > input.run.createdAt,
|
||||
);
|
||||
const exhausted = input.coordinator.attempt >= 3;
|
||||
const phase = supersededByNewerRun || exhausted
|
||||
? "terminal_failure" as const
|
||||
: "retryable_failure" as const;
|
||||
const priorFailureDetail = record(input.coordinator.failureDetail);
|
||||
const workspaceFinalizeAttempt =
|
||||
input.failureScope === "workspace"
|
||||
? (typeof priorFailureDetail.workspaceFinalizeAttempt === "number" &&
|
||||
Number.isInteger(priorFailureDetail.workspaceFinalizeAttempt) &&
|
||||
priorFailureDetail.workspaceFinalizeAttempt >= 0
|
||||
? priorFailureDetail.workspaceFinalizeAttempt
|
||||
: 0) + 1
|
||||
: null;
|
||||
const exhausted =
|
||||
input.permanent === true ||
|
||||
(workspaceFinalizeAttempt !== null
|
||||
? workspaceFinalizeAttempt >= 3
|
||||
: input.coordinator.attempt >= 3);
|
||||
const phase =
|
||||
supersededByNewerRun || exhausted
|
||||
? ("terminal_failure" as const)
|
||||
: ("retryable_failure" as const);
|
||||
const failureCode = supersededByNewerRun
|
||||
? "native_finalization_superseded"
|
||||
: exhausted
|
||||
? "native_finalization_retry_exhausted"
|
||||
: input.failureCode;
|
||||
await tx.update(nativeRunFinalizations).set({
|
||||
phase,
|
||||
leaseOwner: null,
|
||||
leaseExpiresAt: null,
|
||||
failureCode,
|
||||
failureDetail: {
|
||||
message: input.message.slice(0, 2_000),
|
||||
originalFailureCode: input.failureCode,
|
||||
recoveryOwner: supersededByNewerRun
|
||||
? { kind: "none", reason: "newer_native_decision" }
|
||||
: exhausted
|
||||
? { kind: "board" }
|
||||
: { kind: "agent", agentId: input.run.agentId },
|
||||
nextAction: input.nextAction,
|
||||
},
|
||||
nextAttemptAt: supersededByNewerRun || exhausted ? null : nextAttemptAt,
|
||||
updatedAt: now,
|
||||
}).where(eq(nativeRunFinalizations.runId, input.run.id));
|
||||
const projectsTerminalStatus = exhausted && !supersededByNewerRun && input.projectRunStatus;
|
||||
const [updatedRun] = await tx.update(heartbeatRuns).set({
|
||||
...(projectsTerminalStatus ? {
|
||||
status: exhaustedRunStatus,
|
||||
finishedAt: input.run.finishedAt ?? now,
|
||||
} : {}),
|
||||
nativePhase: phase,
|
||||
nativePhaseUpdatedAt: now,
|
||||
resultJson: {
|
||||
...record(input.run.resultJson),
|
||||
finalizationPhase: phase,
|
||||
: input.permanent
|
||||
? input.failureCode
|
||||
: exhausted
|
||||
? input.failureScope === "workspace"
|
||||
? "native_workspace_sync_out_retry_exhausted"
|
||||
: "native_finalization_retry_exhausted"
|
||||
: input.failureCode;
|
||||
await tx
|
||||
.update(nativeRunFinalizations)
|
||||
.set({
|
||||
phase,
|
||||
leaseOwner: null,
|
||||
leaseExpiresAt: null,
|
||||
failureCode,
|
||||
originalFailureCode: input.failureCode,
|
||||
nextAttemptAt: supersededByNewerRun || exhausted ? null : nextAttemptAt.toISOString(),
|
||||
},
|
||||
updatedAt: now,
|
||||
}).where(eq(heartbeatRuns.id, input.run.id)).returning();
|
||||
failureDetail: {
|
||||
message: input.message.slice(0, 2_000),
|
||||
originalFailureCode: input.failureCode,
|
||||
...(workspaceFinalizeAttempt === null
|
||||
? {}
|
||||
: { workspaceFinalizeAttempt }),
|
||||
recoveryOwner: supersededByNewerRun
|
||||
? { kind: "none", reason: "newer_native_decision" }
|
||||
: exhausted
|
||||
? { kind: "board" }
|
||||
: { kind: "agent", agentId: input.run.agentId },
|
||||
nextAction: input.nextAction,
|
||||
},
|
||||
nextAttemptAt: supersededByNewerRun || exhausted ? null : nextAttemptAt,
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(eq(nativeRunFinalizations.runId, input.run.id));
|
||||
const projectsTerminalStatus =
|
||||
exhausted && !supersededByNewerRun && input.projectRunStatus;
|
||||
const [updatedRun] = await tx
|
||||
.update(heartbeatRuns)
|
||||
.set({
|
||||
...(projectsTerminalStatus
|
||||
? {
|
||||
status:
|
||||
input.failureScope === "workspace"
|
||||
? "failed"
|
||||
: exhaustedRunStatus,
|
||||
finishedAt: input.run.finishedAt ?? now,
|
||||
}
|
||||
: {}),
|
||||
nativePhase: phase,
|
||||
nativePhaseUpdatedAt: now,
|
||||
resultJson: {
|
||||
...record(input.run.resultJson),
|
||||
finalizationPhase: phase,
|
||||
failureCode,
|
||||
originalFailureCode: input.failureCode,
|
||||
nextAttemptAt:
|
||||
supersededByNewerRun || exhausted
|
||||
? null
|
||||
: nextAttemptAt.toISOString(),
|
||||
},
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(eq(heartbeatRuns.id, input.run.id))
|
||||
.returning();
|
||||
if (projectsTerminalStatus) terminalRunToEmit = updatedRun ?? null;
|
||||
if (supersededByNewerRun) {
|
||||
await issueRecoveryActionService(tx as unknown as Db).resolveActiveForIssue({
|
||||
@@ -284,9 +337,16 @@ async function recordRetryableFailure(input: {
|
||||
resolutionNote: "A newer native run already committed the authoritative issue decision; the stale finalizer was retired without changing issue state.",
|
||||
}, tx);
|
||||
} else if (exhausted) {
|
||||
await issueService(tx as unknown as Db).update(input.coordinator.issueId, {
|
||||
status: "in_review",
|
||||
}, tx);
|
||||
await issueService(tx as unknown as Db).update(
|
||||
input.coordinator.issueId,
|
||||
{
|
||||
status:
|
||||
input.permanent && input.failureScope === "workspace"
|
||||
? "blocked"
|
||||
: "in_review",
|
||||
},
|
||||
tx,
|
||||
);
|
||||
}
|
||||
if (!supersededByNewerRun) {
|
||||
await issueRecoveryActionService(tx as unknown as Db).upsertSourceScoped({
|
||||
@@ -301,10 +361,15 @@ async function recordRetryableFailure(input: {
|
||||
evidence: {
|
||||
runId: input.run.id,
|
||||
coordinatorAttempt: input.coordinator.attempt,
|
||||
...(workspaceFinalizeAttempt === null
|
||||
? {}
|
||||
: { workspaceFinalizeAttempt }),
|
||||
originalFailureCode: input.failureCode,
|
||||
},
|
||||
nextAction: exhausted
|
||||
? `Finalization retry budget exhausted. ${input.nextAction}`
|
||||
? input.permanent
|
||||
? input.nextAction
|
||||
: `Finalization retry budget exhausted. ${input.nextAction}`
|
||||
: input.nextAction,
|
||||
wakePolicy: exhausted
|
||||
? null
|
||||
@@ -328,6 +393,8 @@ export async function recordNativeFinalizationFailure(input: {
|
||||
runId: string;
|
||||
error: unknown;
|
||||
projectRunStatus?: boolean;
|
||||
failureScope?: "provider" | "workspace";
|
||||
permanent?: boolean;
|
||||
}) {
|
||||
const [run, coordinator] = await Promise.all([
|
||||
input.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId))
|
||||
@@ -344,8 +411,15 @@ export async function recordNativeFinalizationFailure(input: {
|
||||
coordinator,
|
||||
failureCode,
|
||||
message,
|
||||
nextAction: "Repair the persisted native result or contract discriminator, then resume finalization from the coordinator.",
|
||||
nextAction:
|
||||
input.failureScope === "workspace"
|
||||
? input.permanent
|
||||
? "Restore the exact sandbox containing the unexported workspace changes, or resolve the task manually from durable evidence."
|
||||
: "Retry workspace export and merge from the retained sandbox; do not submit another provider turn."
|
||||
: "Repair the persisted native result or contract discriminator, then resume finalization from the coordinator.",
|
||||
projectRunStatus: input.projectRunStatus,
|
||||
failureScope: input.failureScope,
|
||||
permanent: input.permanent,
|
||||
});
|
||||
}
|
||||
|
||||
@@ -395,12 +469,23 @@ export async function finalizeNativeRun(input: {
|
||||
workspaceFinalizeStatus: "succeeded" | "failed";
|
||||
/** Reconciliation owns terminal run projection; the live heartbeat does it afterward. */
|
||||
projectRunStatus?: boolean;
|
||||
/** Workspace-only replay must not consume the provider recovery budget. */
|
||||
preserveProviderAttempt?: boolean;
|
||||
failpoint?: NativeStatusCommitFailpoint;
|
||||
}) {
|
||||
const run = await input.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId))
|
||||
.limit(1).then((rows) => rows[0] ?? null);
|
||||
if (!run || run.runtimeMode !== "native") throw new Error("native_finalization_run_missing");
|
||||
const claim = await claimCoordinator({ db: input.db, runId: input.runId });
|
||||
const run = await input.db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.id, input.runId))
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!run || run.runtimeMode !== "native")
|
||||
throw new Error("native_finalization_run_missing");
|
||||
const claim = await claimCoordinator({
|
||||
db: input.db,
|
||||
runId: input.runId,
|
||||
preserveProviderAttempt: input.preserveProviderAttempt,
|
||||
});
|
||||
const coordinator = claim.coordinator;
|
||||
if (!claim.leaseOwner && coordinator.phase === "committed") {
|
||||
if (input.projectRunStatus) await projectCommittedRun({ db: input.db, run, coordinator });
|
||||
|
||||
@@ -2880,7 +2880,7 @@ describe("native warm session supervision", () => {
|
||||
);
|
||||
});
|
||||
|
||||
it("rehydrates a runnerd warm session from its checkpoint under a fresh run authority", async () => {
|
||||
it("reattaches a live runnerd warm session under a fresh run authority", async () => {
|
||||
const stateBase = await mkdtemp(
|
||||
join(tmpdir(), "paperclip-runnerd-warm-authority-"),
|
||||
);
|
||||
@@ -2889,9 +2889,7 @@ describe("native warm session supervision", () => {
|
||||
process.env.PAPERCLIP_RUNNER_STATE_DIR = stateBase;
|
||||
process.env.PAPERCLIP_HOME = stateBase;
|
||||
const firstClose = vi.fn(async () => undefined);
|
||||
const secondClose = vi.fn(async () => undefined);
|
||||
const firstSession = { close: firstClose };
|
||||
const secondSession = { close: secondClose };
|
||||
const first = {
|
||||
...execution,
|
||||
binding: {
|
||||
@@ -2947,18 +2945,8 @@ describe("native warm session supervision", () => {
|
||||
return result;
|
||||
})
|
||||
.mockImplementationOnce(async (options) => {
|
||||
expect(options.existingSession).toBeUndefined();
|
||||
expect(options.persistedSession).toEqual(
|
||||
expect.objectContaining({
|
||||
identity: expect.objectContaining({
|
||||
runId: second.binding.runId,
|
||||
sessionId: second.session.normalizedSessionId,
|
||||
}),
|
||||
providerSessionId: "provider-runnerd-warm",
|
||||
activeTurnId: null,
|
||||
}),
|
||||
);
|
||||
options.onSession?.(secondSession);
|
||||
expect(options.existingSession).toBe(firstSession);
|
||||
expect(options.persistedSession).toBeUndefined();
|
||||
return result;
|
||||
});
|
||||
|
||||
@@ -3014,10 +3002,10 @@ describe("native warm session supervision", () => {
|
||||
runnerInstanceId: "runner-runnerd-warm",
|
||||
useRunnerd: true,
|
||||
});
|
||||
expect(firstClose).toHaveBeenCalledWith({
|
||||
reason: "warm native session authority epoch rotated",
|
||||
});
|
||||
await vi.waitFor(() => expect(secondClose).toHaveBeenCalled(), {
|
||||
expect(firstClose).not.toHaveBeenCalled();
|
||||
await vi.waitFor(() => expect(firstClose).toHaveBeenCalledWith({
|
||||
reason: "warm native session idle timeout",
|
||||
}), {
|
||||
timeout: 500,
|
||||
});
|
||||
} finally {
|
||||
@@ -5392,9 +5380,13 @@ describe("runnerd provider runtime wiring", () => {
|
||||
|
||||
const firstOptions = state.createBackend.mock.calls[0]![1];
|
||||
const continuationOptions = state.createBackend.mock.calls[1]![1];
|
||||
await expect(firstOptions.dynamicToolHandler!({})).rejects.toThrow(
|
||||
"native_tool_authority_epoch_revoked",
|
||||
);
|
||||
// The retained runner backend owns one stable callback. After run.attach,
|
||||
// that callback routes through the session-scope authority registry to
|
||||
// the new run; stale provider calls are rejected earlier by runnerd's
|
||||
// turn identity boundary.
|
||||
await expect(firstOptions.dynamicToolHandler!({})).resolves.toEqual({
|
||||
runId: continuation.binding.runId,
|
||||
});
|
||||
await expect(
|
||||
continuationOptions.dynamicToolHandler!({}),
|
||||
).resolves.toEqual({ runId: continuation.binding.runId });
|
||||
|
||||
@@ -2623,7 +2623,13 @@ export async function nativeProviderRecoveryEvidence(input: {
|
||||
const durableEvents = await input.db
|
||||
.select({ eventType: heartbeatRunEvents.eventType })
|
||||
.from(heartbeatRunEvents)
|
||||
.where(eq(heartbeatRunEvents.runId, input.runId));
|
||||
.where(
|
||||
and(
|
||||
eq(heartbeatRunEvents.runId, input.runId),
|
||||
inArray(heartbeatRunEvents.eventType, [...PROVIDER_DURABLE_EVENT_TYPES]),
|
||||
),
|
||||
)
|
||||
.limit(1);
|
||||
const providerEventsExist = durableEvents.some((event) =>
|
||||
PROVIDER_DURABLE_EVENT_TYPES.has(event.eventType),
|
||||
);
|
||||
@@ -4171,23 +4177,7 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
entry.busy = true;
|
||||
if (entry.idleTimer !== null) clearTimeout(entry.idleTimer);
|
||||
entry.idleTimer = null;
|
||||
if (input.useRunnerd) {
|
||||
// A retained driver closes over the tool authority from the run that
|
||||
// created it. Preserve provider continuity through its durable
|
||||
// checkpoint, but rebuild the runnerd backend so the next run gets a
|
||||
// fresh, immutable authority epoch. Reusing the live driver would
|
||||
// either retain stale authority or require rebinding old callbacks.
|
||||
warmNativeSessions.delete(warmSessionId);
|
||||
await entry.session.close({
|
||||
reason: "warm native session authority epoch rotated",
|
||||
});
|
||||
persistedWarmSession = loadWarmNativeCheckpoint(
|
||||
input.execution,
|
||||
warmConfigDigest,
|
||||
);
|
||||
} else {
|
||||
existingWarmSession = entry.session;
|
||||
}
|
||||
existingWarmSession = entry.session;
|
||||
}
|
||||
} else {
|
||||
persistedWarmSession = loadWarmNativeCheckpoint(
|
||||
@@ -7235,6 +7225,13 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
input.restartRecovery?.kind === "reattach_existing_runner"
|
||||
? input.restartRecovery.process
|
||||
: null;
|
||||
const executeCurrentToolAuthority = (
|
||||
call: Parameters<SessionToolAuthorityEpoch["execute"]>[0],
|
||||
) => {
|
||||
const current = sessionToolAuthorityEpochs.get(sessionScopeId);
|
||||
if (!current) throw new Error("native_session_tool_authority_unavailable");
|
||||
return current.execute(call);
|
||||
};
|
||||
const backend = createNativeSessionBackend(runnerExecution, {
|
||||
runnerInstanceId: input.runnerInstanceId,
|
||||
environment: effectiveRunnerEnvironment,
|
||||
@@ -7243,8 +7240,8 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
: "local_filesystem",
|
||||
onSpawn: input.onSpawn,
|
||||
dynamicTools,
|
||||
dynamicToolHandler: (call) => authorityEpoch.execute(call),
|
||||
acpxDynamicToolHandler: (call) => authorityEpoch.execute(call),
|
||||
dynamicToolHandler: executeCurrentToolAuthority,
|
||||
acpxDynamicToolHandler: executeCurrentToolAuthority,
|
||||
opencodeRuntimeDirectory: resolve(
|
||||
resolvePaperclipInstanceRoot(),
|
||||
"runtime",
|
||||
@@ -7419,7 +7416,7 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
turnId: `turn-${input.execution.binding.runId}`,
|
||||
itemId: `item-${input.execution.binding.runId}`,
|
||||
},
|
||||
controlPlaneRegistration: (authority) =>
|
||||
controlPlaneRegistration: (authority, attachmentIdentity) =>
|
||||
measureNativeRunnerSpan(
|
||||
input.trace,
|
||||
"runner.transport.connect",
|
||||
@@ -7446,7 +7443,9 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
() =>
|
||||
registerRunnerPrpAuthority({
|
||||
companyId: input.execution.binding.companyId,
|
||||
runId: input.execution.binding.runId,
|
||||
runId:
|
||||
attachmentIdentity?.runId ??
|
||||
input.execution.binding.runId,
|
||||
authority,
|
||||
}),
|
||||
);
|
||||
@@ -7469,7 +7468,9 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
() =>
|
||||
resolvePaperclipRunnerTransport({
|
||||
target,
|
||||
runId: input.execution.binding.runId,
|
||||
runId:
|
||||
attachmentIdentity?.runId ??
|
||||
input.execution.binding.runId,
|
||||
localConnectUrl: "ws://127.0.0.1/unused",
|
||||
runnerPublicUrl: input.runnerPublicUrl,
|
||||
runnerCaBundlePath: input.runnerCaBundlePath,
|
||||
@@ -7505,7 +7506,9 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
() =>
|
||||
resolvePaperclipRunnerTransport({
|
||||
target,
|
||||
runId: input.execution.binding.runId,
|
||||
runId:
|
||||
attachmentIdentity?.runId ??
|
||||
input.execution.binding.runId,
|
||||
localConnectUrl: "ws://127.0.0.1/unused",
|
||||
runnerPublicUrl: input.runnerPublicUrl,
|
||||
runnerCaBundlePath: input.runnerCaBundlePath,
|
||||
@@ -7541,7 +7544,9 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
() =>
|
||||
registerRunnerPrpAuthority({
|
||||
companyId: input.execution.binding.companyId,
|
||||
runId: input.execution.binding.runId,
|
||||
runId:
|
||||
attachmentIdentity?.runId ??
|
||||
input.execution.binding.runId,
|
||||
authority,
|
||||
}),
|
||||
{ parentName: "runner.transport.connect" },
|
||||
|
||||
@@ -10,6 +10,13 @@ import {
|
||||
} from "@paperclipai/db";
|
||||
import { workspaceOperationService } from "../workspace-operations.js";
|
||||
import { inspectManagedGitWorktreeBranch } from "../workspace-runtime.js";
|
||||
import { environmentService } from "../environments.js";
|
||||
import type { EnvironmentRuntimeService } from "../environment-runtime.js";
|
||||
import { resolveEnvironmentExecutionTarget } from "../environment-execution-target.js";
|
||||
import {
|
||||
readNativeWorkspaceSyncReference,
|
||||
resumeNativeWorkspaceSync,
|
||||
} from "./native-workspace-sync.js";
|
||||
|
||||
function record(value: unknown): Record<string, unknown> {
|
||||
return value && typeof value === "object" && !Array.isArray(value)
|
||||
@@ -21,14 +28,27 @@ function readString(value: unknown) {
|
||||
return typeof value === "string" && value.trim().length > 0 ? value.trim() : null;
|
||||
}
|
||||
|
||||
function workspaceSyncFailure(
|
||||
code: "workspace_sync_out_failed" | "workspace_sync_out_unrecoverable",
|
||||
) {
|
||||
return {
|
||||
status: "failed" as const,
|
||||
exitCode: 1,
|
||||
stderr: `${code}\n`,
|
||||
metadata: { workspaceSync: { code } },
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Resume the real workspace-finalization action for a result-bearing native
|
||||
* run. This service observes the workspace; it never fabricates a successful
|
||||
* marker. The caller decides whether a failed observation is retryable.
|
||||
* run. Durable sandbox-backed runs export and merge the remote workspace;
|
||||
* older runs retain the local workspace validation path. The caller owns the
|
||||
* durable retry and failure policy.
|
||||
*/
|
||||
export async function resumeNativeWorkspaceFinalization(input: {
|
||||
db: Db;
|
||||
runId: string;
|
||||
environmentRuntime?: EnvironmentRuntimeService;
|
||||
}) {
|
||||
const bound = await input.db.select({
|
||||
companyId: heartbeatRuns.companyId,
|
||||
@@ -57,7 +77,12 @@ export async function resumeNativeWorkspaceFinalization(input: {
|
||||
eq(workspaceOperations.phase, "workspace_finalize"),
|
||||
)).orderBy(desc(workspaceOperations.createdAt)).limit(1).then((rows) => rows[0] ?? null);
|
||||
|
||||
const persistedInput = record(record(bound.runnerProfileJson).nativeExecutionInput);
|
||||
const persistedInput = record(
|
||||
record(bound.runnerProfileJson).nativeExecutionInput,
|
||||
);
|
||||
const nativeWorkspaceSync = readNativeWorkspaceSyncReference(
|
||||
record(bound.runnerProfileJson).nativeWorkspaceSync,
|
||||
);
|
||||
const binding = record(persistedInput.binding);
|
||||
const workspaceId = previous?.executionWorkspaceId
|
||||
?? bound.issueWorkspaceId
|
||||
@@ -87,6 +112,88 @@ export async function resumeNativeWorkspaceFinalization(input: {
|
||||
: "workspace_directory",
|
||||
},
|
||||
run: async () => {
|
||||
if (nativeWorkspaceSync) {
|
||||
if (!input.environmentRuntime) {
|
||||
return {
|
||||
status: "failed",
|
||||
exitCode: 1,
|
||||
stderr:
|
||||
"Native workspace finalization cannot access the environment runtime.\n",
|
||||
};
|
||||
}
|
||||
const environmentsSvc = environmentService(input.db);
|
||||
const lease = await environmentsSvc.getLeaseById(
|
||||
nativeWorkspaceSync.leaseId,
|
||||
);
|
||||
const environment = lease?.environmentId
|
||||
? await environmentsSvc.getById(lease.environmentId)
|
||||
: null;
|
||||
if (
|
||||
!lease ||
|
||||
!environment ||
|
||||
lease.companyId !== bound.companyId ||
|
||||
environment.id !== lease.environmentId
|
||||
) {
|
||||
return workspaceSyncFailure("workspace_sync_out_unrecoverable");
|
||||
}
|
||||
if (
|
||||
lease.status === "expired" ||
|
||||
lease.status === "failed" ||
|
||||
lease.status === "pending_cleanup" ||
|
||||
!lease.providerLeaseId ||
|
||||
lease.providerLeaseId !== nativeWorkspaceSync.providerLeaseId
|
||||
) {
|
||||
return workspaceSyncFailure("workspace_sync_out_unrecoverable");
|
||||
}
|
||||
const target = await resolveEnvironmentExecutionTarget({
|
||||
db: input.db,
|
||||
companyId: bound.companyId,
|
||||
adapterType: "paperclip_runner",
|
||||
environment,
|
||||
leaseId: lease.id,
|
||||
leaseMetadata: lease.metadata,
|
||||
lease,
|
||||
environmentRuntime: input.environmentRuntime,
|
||||
});
|
||||
if (!target) {
|
||||
return {
|
||||
status: "failed",
|
||||
exitCode: 1,
|
||||
stderr: "Native workspace finalization target is unavailable.\n",
|
||||
};
|
||||
}
|
||||
try {
|
||||
const restored = await resumeNativeWorkspaceSync({
|
||||
db: input.db,
|
||||
runId: input.runId,
|
||||
target,
|
||||
});
|
||||
if (!restored) {
|
||||
return workspaceSyncFailure("workspace_sync_out_unrecoverable");
|
||||
}
|
||||
return {
|
||||
status: "succeeded",
|
||||
exitCode: 0,
|
||||
system:
|
||||
"Native workspace finalization restored the remote workspace.\n",
|
||||
metadata: {
|
||||
workspaceSync: {
|
||||
schema: nativeWorkspaceSync.schema,
|
||||
workspaceId: nativeWorkspaceSync.workspaceId,
|
||||
leaseId: nativeWorkspaceSync.leaseId,
|
||||
},
|
||||
},
|
||||
};
|
||||
} catch (error) {
|
||||
const code =
|
||||
error instanceof Error &&
|
||||
(error.message === "workspace_sync_out_unrecoverable" ||
|
||||
error.message.includes("daytona_sandbox_not_found"))
|
||||
? "workspace_sync_out_unrecoverable"
|
||||
: "workspace_sync_out_failed";
|
||||
return workspaceSyncFailure(code);
|
||||
}
|
||||
}
|
||||
if (!cwd) {
|
||||
return {
|
||||
status: "failed",
|
||||
|
||||
File diff suppressed because it is too large.
Load diff
Reference in new issue
Block a user