Jannik/PaperClipAI
1
0 Fork 0
mirror of https://github.com/paperclipai/paperclip.git synced 2026-11-08 22:13:59 +01:00
Code Issues Packages Projects Releases Wiki Activity

fix(runtime): reclaim failed remote checkpoint retries (#18877)

Browse Source 
## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Native recovery must preserve the original execution and provider
state.
> - A failed remote run can retain its provider PID after heartbeat
releases the task execution lock.
> - Normal resumption excludes that PID, while restart recovery rejects
the released lock.
> - Recovery must reclaim that exact execution only after current
admission and durable ownership checks pass.
> - Failed retries and admission holds need separate finite budgets.
Neither can prove a provider stopped.

## Linked Issues or Issue Description

**What happened?**

A failed remote native run remains in retryable failure without a task
execution owner. Normal resumption and restart recovery both reject it
before inspecting the preserved remote provider. The current remote
workspace reference format also misses the due-retry routing path.

**Expected behavior**

A due exact-checkpoint retry reclaims its original task execution only
when current ownership and safety gates permit it. Newer task input
prevents an old claim. Lost or ambiguous provider state still prevents
replay or replacement. Ineligible recovery eventually becomes a quiet
manual reconciliation obligation.

**Steps to reproduce**

1. Start a remote native run and persist its provider checkpoint and
encrypted Runner authority.
2. Fail with exact-checkpoint recovery and a retained remote PID.
3. Release the task execution lock at heartbeat completion.
4. Run ordinary resumption and restart recovery after the retry
deadline.
5. Observe that both reject the same eligible run before remote
inspection.

**Paperclip version or commit**

Reproduced from source at `d59791828482`; integrated with upstream
`db9be4c5a6`.

**Deployment mode**

Remote workspaces with PostgreSQL. Local regressions use disposable
PostgreSQL and encrypted checkpoint persistence.

Related public repairs #18846, #18854, #18860 and #18841 cover finished
per-turn owners, pre-provider failures, unstarted follow-ups and storage
migration. They do not reclaim this failed run's released execution
lock. Merged #18879 also adds scheduled-resume lock retention and
reclaim. This PR keeps that compatibility and requires encrypted Runner
authority, canonical source ordering and a charged failed retry before
reclaim. Its scheduled-resume regression now uses a real encrypted
authority and retains a snapshot-only refusal control.

## What Changed

- Share normal admission checks and one guarded execution-binding
restore/event helper between normal and remote resumption.
- Keep agent-name normalization in a module with no imports. Preserve
heartbeat's existing export. Native admission no longer loads heartbeat
orchestration for a string key.
- Recheck company, assignment, agent eligibility, Stop, pause, approval,
dependencies, execution and checkout ownership before reclaim.
- Reject newer task input using PostgreSQL creation-time and ID
ordering, preserving timestamp precision.
- Require the recorded remote target, encrypted Runner authority, exact
checkpoint identity and existing controller fence.
- Commit task binding, controller lease, run status and recovery event
atomically. Recognize both remote workspace reference formats.
- Charge failed reattachments against the existing
three-provider-attempt budget, including retained-lock crash windows.
Active controller adoption preserves its charged attempt.
- Use the existing five-retry/30-minute system-recovery policy for
admission holds. Ended tasks and newer sources exhaust it immediately.
Preserve PID, checkpoint and task bindings, suppress resource
settlement, and assign manual reconciliation with executor stop/state
explicitly unverified.
- Register recovery intent before deadline writes; preserve live/future
deadlines quietly and roll back zero-row writes. Record changed hold
reasons once and terminal dispositions separately.
- Keep historical provider-budget exhaustion behind current
admission/source/controller/authority checks. Make an existing
orphan-cleanup fixture explicitly due.
- Include exact plugin task leases that have no workspace reference.
Reuse every reclaim and retry-budget guard, with a bounded
company/task/run lease projection.
- Share strict manual ownership proof across stale-lock cleanup,
adoption, queue release and wake admission. Require termination receipts
for every exact remote lease before normal cleanup can release either
binding.
- Preserve governing Stop through the existing audited cancellation
path. Its acknowledgement does not release an unverified provider.
Ordinary Retry can wait for exact Stop and file preservation, then queue
one continuation.

## Verification

- Current head is `5d7baf74aa89d00d7ac6e1493d38a6ce2ae7b7b5`. Its
[current CI
matrix](https://github.com/paperclipai/paperclip/actions/runs/41782198858)
is terminal and clean: 54 successful, three skipped and two neutral
contexts. Required `ci / verify` and `ci / e2e` pass. Greptile is 5/5 on
this exact head with no open findings. No fresh Paperclip 5/5 assessment
is claimed.
- The [prior complete CI
matrix](https://github.com/paperclipai/paperclip/actions/runs/41754365388)
passed on `3e641466d3edcaa6dd5021d979627082d388eb05`, with Greptile and
Paperclip both 5/5 and no open threads.
- All 233 tests across the six affected heartbeat, restart-isolation,
remote-admission, scheduled-resume and session-binding families pass on
production-identical `a5104fa89f9cb493091e000bf21ed3858871ae46` under
Node 24.22.2 and pnpm 9.15.4. They cover both workspace formats and
plugin task leases, competing controllers, exact rollback, Stop,
assignment/company boundaries, attempts, work-intent ordering, finite
backoff and quiet held state.
- The merge preserves #18931's per-run audit-failure isolation and this
PR's unchanged-transition hold. Two initial composition cases failed
because their setup or expected result described the older contract. The
rollback test now expects the exact per-run deferral while retaining
full coordinator equality and no work intent. The routine fixture writes
a real encrypted checkpoint through the normal owner before reaching the
actual routine unique constraint. No production admission guard or
rollback assertion was weakened. A separate agent reviewed this exact
merge and fixture delta without finding a blocker.
- Full recursive `pnpm -r typecheck` and `pnpm build` pass on production
merge `89c616fe45ceff75db450f67fd34224b4df7441d`. Subsequent commits
change only three test files. Both the two-file `a5104fa` and final
`5d7baf` server test-layer typechecks pass.
- Four production-identical `a5104fa` probes pass with the rebuilt
actual Rust Runner, a scripted ACPX subprocess and encrypted PostgreSQL
checkpoints under Node 24.22.2. Workspace-reference and plugin-lease
admission each cover surviving and lost providers. Replacement preserves
one provider launch, one original side effect, the charged attempt and
restored task execution binding. Lost providers reject snapshot recovery
without replacement. This is protocol-boundary evidence, not a live
Claude or hosted provisioning claim.
- Initial `a5104fa` CI found one additional routine integration fixture
missing the persisted `exact_checkpoint_resume` mode. Its encrypted
checkpoint was already present. The exact case fails locally before
repair and passes after recording that mode and expecting the charged
provider attempt. Every exact routine conflict, rollback, checkpoint,
task binding, successor-count and sibling-startup assertion remains.
Final head changes only that fixture. The correct native-server group
passes all nine files: 175 executed cases and 14 container-only skips,
matching the now-successful native-server CI gate.
- The earlier canonical HTTP Retry acceptance passed on the prior
reviewed head with real PostgreSQL, the recovery worker and real files.
Pending Stop protects both bindings and starts no successor. Exact
termination and copyback queue one preserved-workspace continuation. No
browser or live-model outcome is inferred from that test.
- The predecessor's full local `pnpm test:run` failed in an unchanged
Discord Gateway fixture with a 200 ms lease and a 422 initial ownership
error. Focused predecessor and untouched-base controls pass; that
full-run cause remains unproved. The required command was also started
on production-identical `a5104fa`. All six local chat shards passed,
covering 1,850 cases. The duplicate local invocation was interrupted
with SIGINT while the general-server group was still running after this
head's complete CI and the combined repair qualification had passed. The
wrapper exited 1 on cancellation, no process retained its test log, and
no full local pass is claimed. An earlier attempt to select the native
lane through the CI wrapper instead selected local package sharding:
2,631 cases passed and 13 failed. Those separate failures are retained
without causal attribution or unrelated source changes; that run is not
counted as native-server qualification. The direct native group and
current complete CI matrix pass.
- Live Claude Opus 5.5 continuation, hosted plugin provisioning,
deployment and recovery of historical tasks remain unverified. No
original task was retried or mutated for this verification.

## Risks

- Restoring a task binding grants remote inspection; it does not prove
that the provider survived.
- Unknown effects, lost providers and missing semantic results retain
their existing reconciliation requirements.
- Exhausted admission holds preserve execution bindings and resource
evidence for manual reconciliation. They do not authorize a fresh
executor or provider/resource release.
- The source guard is conservative: newer task input prevents recovery
of an older failed source.
- No database migration or provider profile change is required.

## Model Used

OpenAI Codex, based on GPT-6, with reasoning, tool use, code editing and
shell execution. The runtime does not expose the exact model snapshot or
context window size.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed 2026-11-04 07:18:07 -06:00
1 parent c39e076638
commit d8c1b9d4b5
24 files changed
+1334 -166
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

server/src/services/native-runtime/native-cancellation-request.ts
+4 -1
View File
@@ -2,6 +2,7 @@ import { notifyScheduledRetryCancellation } from "../scheduled-retry-cancellatio
import { and, eq, sql } from "drizzle-orm";
import { activityLog, budgetReservations, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db";
import { badRequest, conflict } from "../../errors.js";
import { nativeManualAdmissionHoldMatches } from "./native-runner-ownership.js";
const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
const record = (value: unknown): Record<string, unknown> => value !== null && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : {};
@@ -78,8 +79,10 @@ export function assertCancellationRequest(result: unknown, requestId: string, al
* is not retry authority; a resumed Stop needs its exact durable native intent. */
export function nativeRetryCancellationEligible(input: {
runId: string; companyId: string; issueId: string; resultJson: unknown; requestId?: string; scope?: string;
}, coordinator: { phase?: string; failureCode?: string | null } | null | undefined) {
}, coordinator: Partial<Pick<typeof nativeRunFinalizations.$inferSelect,
"runId" | "companyId" | "issueId" | "phase" | "resultId" | "failureCode" | "failureDetail">> | null | undefined) {
if (coordinator?.phase === "retryable_failure") return true;
if (coordinator?.failureCode === "native_restart_recovery_blocked" && nativeManualAdmissionHoldMatches(input, coordinator)) return true;
if (coordinator?.phase !== "terminal_failure" || coordinator.failureCode !== "native_retry_cancelled") return false;
const result = record(input.resultJson), native = record(result.nativeCancellation);
return (!input.requestId || (record(result.startupCancellation).cancellationRequestId === input.requestId
server/src/services/native-runtime/native-finalization-reconciler.ts
+5 -47
View File
@@ -1,5 +1,4 @@
import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js";
import { notifyStatusCardIssueChange } from "../status-card-notifications.js";
import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js";
import { pendingNativeRunProjectionCondition } from "./native-run-projection.js";
import { nativeStatusReassessmentSourceIsSuperseded, reassessNativeStatusFromSource } from "./native-status-reassessment.js";
@@ -15,10 +14,9 @@ import { dismissAutomaticCompletionReviews } from "./automatic-completion-review
import { logger } from "../../middleware/logger.js";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
import { createHash, randomUUID } from "node:crypto";
import { and, asc, desc, eq, gt, inArray, isNotNull, isNull, lt, lte, ne, not, notInArray, or, sql, type SQL } from "drizzle-orm";
import { and, asc, desc, eq, gt, inArray, isNotNull, isNull, lt, lte, not, notInArray, or, sql, type SQL } from "drizzle-orm";
import type { Db } from "@paperclipai/db";
import {
agents,
heartbeatRunEvents,
heartbeatRuns,
issues,
@@ -57,11 +55,7 @@ import {
type NativeAuthoritativeIssueStatus,
} from "./status-arbiter.js";
import { getNativeReviewAssignment } from "./native-review-participant.js";
import { evaluateAgentInvokabilityFromDb } from "../agent-invokability.js";
import { isAutomaticRecoverySuppressedByPauseHold } from "../recovery/pause-hold-guard.js";
import { readRunCancellation } from "../run-cancellation.js";
import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js";
import { readNativeSessionResumptionAgent, restoreNativeSessionExecution } from "./native-session-resumption-admission.js";
import { retireSupersededNativeRetry } from "./native-superseded-retry.js";
import { legacyWorkspaceRecoveryNotMigrated } from "../legacy-workspace-recovery-migration.js";
@@ -180,37 +174,13 @@ export async function claimNativeSessionResumptions(input: {
)
return false;
if ((issue.executionRunId && issue.executionRunId !== run.id) ||
(issue.checkoutRunId && issue.checkoutRunId !== run.id)) return false;
const [agent] = await tx.select().from(agents).where(and(
eq(agents.companyId, candidate.companyId), eq(agents.id, run.agentId),
)).limit(1);
if (!agent || !(await evaluateAgentInvokabilityFromDb(tx as unknown as Db, agent)).invokable) return false;
if (issue.assigneeAgentId !== run.agentId && !await getNativeReviewAssignment(tx as unknown as Db, {
companyId: run.companyId, issueId: issue.id, agentId: run.agentId,
contextSnapshot: run.contextSnapshot,
})) return false;
if (await isAutomaticRecoverySuppressedByPauseHold(tx as unknown as Db, run.companyId, issue.id)) return false;
if (!(await issueService(tx as unknown as Db).getDependencyReadiness(issue.id, tx)).isDependencyReady) return false;
const [competing] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
eq(heartbeatRuns.companyId, run.companyId), ne(heartbeatRuns.id, run.id),
inArray(heartbeatRuns.status, issue.executionRunId === run.id
? ["running", "scheduled_retry"] : ["queued", "running", "scheduled_retry"]),
or(eq(heartbeatRuns.issueId, issue.id), eq(heartbeatRuns.nativeIssueId, issue.id),
sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issue.id}`),
)).limit(1);
if (competing) return false;
const agent = await readNativeSessionResumptionAgent(tx as unknown as Db, { issue, run, coordinator });
if (!agent) return false;
const profile = row.run.runnerProfileJson ?? {};
const persistedInput = profile.nativeExecutionInput;
const checkpoint = profile.sessionCheckpoint;
const failureDetail = record(row.coordinator.failureDetail);
const cancellation = readRunCancellation(run.resultJson);
const nativeCancellation = record(record(run.resultJson).nativeCancellation);
if ((cancellation?.source === "operator" && cancellation.expected) ||
(["run", "issue"].includes(String(nativeCancellation.scope)) &&
["pending", "acknowledged"].includes(String(nativeCancellation.dispatchState))) ||
typeof failureDetail.successorRunId === "string") return false;
const originalFailureCode =
typeof failureDetail.originalFailureCode === "string"
? failureDetail.originalFailureCode
@@ -345,19 +315,7 @@ export async function claimNativeSessionResumptions(input: {
nativePhaseUpdatedAt: now,
updatedAt: now,
}).where(eq(heartbeatRuns.id, row.run.id));
await notifyStatusCardIssueChange(tx, run.companyId);
await tx.update(issues).set({
executionRunId: run.id,
executionAgentNameKey: agent.name.trim().toLowerCase() || null,
executionLockedAt: now,
updatedAt: now,
}).where(and(eq(issues.id, issue.id), eq(issues.companyId, run.companyId)));
if (issue.executionRunId !== run.id) await appendHeartbeatRunEvent(tx as unknown as Db, {
companyId: run.companyId, runId: run.id, agentId: run.agentId,
eventType: "native.recovery.execution_reclaimed", stream: "system", level: "info",
message: "Reclaimed the original task execution for native session recovery.",
payload: { issueId: issue.id },
});
await restoreNativeSessionExecution(tx as unknown as Db, { issue, run, agentName: agent.name, now });
return true;
}).catch((error: unknown) => {
claimFailed = true;
server/src/services/native-runtime/native-remote-retry-admission.test.ts
+785
View File
@@ -0,0 +1,785 @@
import { randomUUID } from "node:crypto";
import { resolve } from "node:path";
import { and, eq, sql } from "drizzle-orm";
import { agents, companies, completionContracts, databaseWorkPending, environmentLeases, reconcileDatabaseWork, heartbeatRunEvents, heartbeatRuns, issues, issueRelations,
issueTreeHolds, issueTreeHoldMembers, issueRecoveryActions, nativeRunFinalizations, nativeRunResults, nativeRunnerCheckpoints } from "@paperclipai/db";
import { expect, it, vi } from "vitest";
import type { NativeExecutionInputV1 } from "@paperclipai/paperclip-runner";
const provider = vi.hoisted(() => ({ execute: vi.fn() }));
vi.mock("../../vendor/paperclip-runner/index.js", async importOriginal => ({
...await importOriginal<typeof import("../../vendor/paperclip-runner/index.js")>(),
executeNativeSession: provider.execute,
}));
import { DurablePrpControlPlane } from "../../vendor/paperclip-runner/index.js";
import { describeEmbeddedPostgres, nativeRestartFixture, remotePreparedWorkspace } from "../../__tests__/helpers/native-restart.js";
import { nativeRunnerCheckpointWriter } from "./native-runner-checkpoint.js";
import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../delivery-work-notifications.js";
import { claimNativeSessionResumptions } from "./native-finalization-reconciler.js";
import { cancelNativeSession, executePaperclipNativeSession } from "./native-session-executor.js";
import { buildNativeCompletionContract } from "./completion-contracts.js";
import { claimNativeRestartRecoveries, currentNativeControllerIdentity, NATIVE_RESTART_RECOVERY_FAILED } from "./native-restart-recovery.js";
import { readSystemRecovery, SYSTEM_RECOVERY_MAX_RETRIES } from "../system-recovery-state.js";
import { recoveryService } from "../recovery/service.js";
import { issueService } from "../issues.js";
import { hasPendingRunCleanup } from "../execution-blocker.js";
import { createPostgresWakeQueueAdapter } from "../../modules/wake-queue/adapters/postgres.js";
import { nativeManualAdmissionHoldsExecution } from "./native-runner-ownership.js";
import { remoteTerminationReceipt } from "../remote-execution-termination.js";
import { releaseSettledNativeRunProcess } from "../native-local-process-stop.js";
import { createHeartbeatQueue } from "../heartbeat/queue.js";
import { createHeartbeatQueueTestDependencies } from "../../__tests__/helpers/heartbeat-queue.js";
import { instanceSettingsService } from "../instance-settings.js";
import { claimCancellationRequest } from "./native-cancellation-request.js";
describeEmbeddedPostgres("remote native retry execution admission", () => {
const environment = nativeRestartFixture();
async function failedRemoteRun(storage: "classic" | "remote" | "plugin") {
const { companyId, agentId, runtimeRoot } = environment.current;
const fixture = await environment.seedRun(`REMOTE-RETRY-${storage}-${randomUUID()}`);
await fixture.db.update(agents).set({ status: "active" }).where(eq(agents.id, agentId));
const controller = { bootId: randomUUID(), generation: 1, leaseOwner: randomUUID() };
await fixture.db.update(nativeRunFinalizations).set({
controllerBootId: controller.bootId, controllerGeneration: controller.generation,
leaseOwner: controller.leaseOwner, leaseExpiresAt: new Date(Date.now() + 60_000),
}).where(eq(nativeRunFinalizations.runId, fixture.runId));
const authority = new DurablePrpControlPlane({
stateDirectory: resolve(runtimeRoot, fixture.runId),
identity: { runId: fixture.runId, normalizedSessionId: fixture.native.normalizedSessionId,
runnerInstanceId: fixture.native.runnerInstanceId, environmentLeaseId: fixture.native.environmentLeaseId,
turnId: fixture.native.turnId, itemId: fixture.native.itemId },
expectedRunnerVersion: "retry-admission-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}`,
});
authority.setCheckpointPersistence(nativeRunnerCheckpointWriter(fixture.db, {
companyId, issueId: fixture.issueId, runId: fixture.runId, authorityRunId: fixture.runId,
...controller,
}));
await authority.flushCheckpoint();
await authority.stop();
const classic = remotePreparedWorkspace();
const remote = { schema: "paperclip.remote-workspace-run.v1", workspaceId: classic.workspaceId,
leaseId: classic.leaseId, providerLeaseId: classic.providerLeaseId, remoteCwd: classic.remoteCwd,
epoch: 1, state: "prepared", snapshotId: null };
const checkpoint = { backendKind: "acpx", driverKind: "acpx_runtime", sessionId: "provider-session",
providerSessionId: "provider-session", activeTurnId: "retained-turn", semanticResult: null,
pendingRuntimeRequests: [], terminalTurns: [], identity: { companyId, agentId,
issueId: fixture.issueId, runId: fixture.runId, sessionId: fixture.native.normalizedSessionId } };
const [pluginLease] = storage === "plugin" ? await fixture.db.insert(environmentLeases).values({ companyId,
issueId: fixture.issueId, heartbeatRunId: fixture.runId, provider: "daytona", providerLeaseId: randomUUID(),
status: "active", leasePolicy: "ephemeral", metadata: { runnerTaskLease: true },
}).returning({ id: environmentLeases.id }) : [];
await fixture.db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date(),
processPid: 1386, processStartedAt: new Date("2026-11-01T12:00:00Z"),
nativePhase: "retryable_failure", errorCode: "native_session_interrupted",
runnerProfileJson: { ...fixture.preparedProfile, nativeExecutionInput: { binding: {
companyId, agentId, issueId: fixture.issueId, runId: fixture.runId } }, sessionCheckpoint: checkpoint,
...(storage === "plugin" ? { nativeWorkspaceSync: null, remoteWorkspace: null }
: storage === "classic" ? { nativeWorkspaceSync: classic } : { remoteWorkspace: remote }) },
}).where(eq(heartbeatRuns.id, fixture.runId));
await fixture.db.update(nativeRunFinalizations).set({ phase: "retryable_failure", attempt: 1,
leaseOwner: null, leaseExpiresAt: null, recoveryState: "resuming_session",
failureCode: "native_session_interrupted", nextAttemptAt: new Date(0),
failureDetail: { recoveryMode: "exact_checkpoint_resume", providerSessionEstablished: true,
checkpointExists: true, providerEventsExist: true },
}).where(eq(nativeRunFinalizations.runId, fixture.runId));
await fixture.db.update(issues).set({ executionRunId: null, checkoutRunId: null })
.where(eq(issues.id, fixture.issueId));
return { ...fixture, companyId, agentId, checkpoint, pluginLeaseId: pluginLease?.id };
}
async function manuallyHeldRemoteRun() {
const f = await failedRemoteRun("remote");
await f.db.update(issues).set({ executionRunId: f.runId, checkoutRunId: f.runId }).where(eq(issues.id, f.issueId));
await f.db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, issueId: f.issueId,
status: "succeeded", createdAt: new Date(Date.now() + 1_000) });
await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor,
restartKind: "hard", runIds: [f.runId] });
return f;
}
it.each(["background sweep", "terminal cleanup", "pending cleanup", "queue release", "agent release"] as const)
("preserves a manual native execution hold through %s", async entry => {
const f = await manuallyHeldRemoteRun();
const [before] = await f.db.select().from(issues).where(eq(issues.id, f.issueId));
if (entry === "background sweep") expect(await recoveryService(f.db, { enqueueWakeup: vi.fn() })
.sweepStaleIssueLocks({ companyId: f.companyId, issueId: f.issueId })).toMatchObject({ cleared: 0 });
if (entry === "terminal cleanup") {
await expect(issueService(f.db).assertCheckoutOwner(f.issueId, f.agentId, randomUUID()))
.rejects.toMatchObject({ status: 409 });
}
if (entry === "agent release") await expect(issueService(f.db).release(f.issueId, f.agentId, randomUUID()))
.rejects.toMatchObject({ status: 409 });
if (entry === "pending cleanup") expect(await hasPendingRunCleanup(f.db, f.companyId, f.runId)).toBe(true);
if (entry === "queue release") await createPostgresWakeQueueAdapter(f.db, {
resolveResponsibleUserId: async () => "operator", getRoutineEnv: async () => ({ routineId: null, env: null, responsibleUserId: null }),
resolveSessionBeforeForWakeup: async () => null,
}).withIssueExecutionLock({ companyId: f.companyId, runId: f.runId, now: new Date() },
async () => ({ outcome: { kind: "released" }, postCommitEffects: [] }));
expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toEqual(before);
expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0])
.toMatchObject({ processPid: 1386, runnerProfileJson: { sessionCheckpoint: f.checkpoint } });
});
it.each(["execution", "checkout"] as const)("preserves a manual hold referenced only by the %s binding", async pointer => {
const f = await manuallyHeldRemoteRun();
await f.db.update(issues).set(pointer === "execution" ? { checkoutRunId: null } : { executionRunId: null })
.where(eq(issues.id, f.issueId));
expect(await recoveryService(f.db, { enqueueWakeup: vi.fn() }).sweepStaleIssueLocks({ companyId: f.companyId, issueId: f.issueId }))
.toMatchObject({ cleared: 0 });
expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0])
.toMatchObject(pointer === "execution" ? { executionRunId: f.runId, checkoutRunId: null } : { executionRunId: null, checkoutRunId: f.runId });
});
it("rechecks a newly committed manual hold under the issue lock before background cleanup", async () => {
const f = await manuallyHeldRemoteRun();
const [hold] = await f.db.select({ failureCode: nativeRunFinalizations.failureCode,
failureDetail: nativeRunFinalizations.failureDetail }).from(nativeRunFinalizations)
.where(eq(nativeRunFinalizations.runId, f.runId));
await f.db.update(nativeRunFinalizations).set({ failureCode: null, failureDetail: null })
.where(eq(nativeRunFinalizations.runId, f.runId));
const beforeStaleLockClear = vi.fn(async (issueId: string) => {
if (issueId === f.issueId) await f.db.update(nativeRunFinalizations).set(hold)
.where(eq(nativeRunFinalizations.runId, f.runId));
});
expect(await recoveryService(f.db, { enqueueWakeup: vi.fn(), beforeStaleLockClear })
.sweepStaleIssueLocks({ companyId: f.companyId, issueId: f.issueId })).toMatchObject({ cleared: 0 });
expect(beforeStaleLockClear).toHaveBeenCalledExactlyOnceWith(f.issueId);
expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0])
.toMatchObject({ executionRunId: f.runId, checkoutRunId: f.runId });
});
it("defers a new wake behind the manual owner without clearing it or dispatching", async () => {
const f = await manuallyHeldRemoteRun();
await f.db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
const [agent] = await f.db.select().from(agents).where(eq(agents.id, f.agentId));
const dependencies = createHeartbeatQueueTestDependencies(f.db);
dependencies.getAgent.mockResolvedValue(agent);
dependencies.instanceSettings.getExperimental.mockImplementation(() => instanceSettingsService(f.db).getExperimental());
dependencies.wakeQueue.admitWakeBehindIssueExecution.mockResolvedValue({ kind: "deferred" });
await createHeartbeatQueue(f.db, dependencies).enqueueWakeup(f.agentId, { source: "assignment", reason: "issue_assigned",
requestedByActorType: "system", payload: { issueId: f.issueId }, contextSnapshot: { issueId: f.issueId } });
expect(dependencies.executeRun).not.toHaveBeenCalled();
expect(dependencies.wakeQueue.admitWakeBehindIssueExecution).toHaveBeenCalledOnce();
expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0])
.toMatchObject({ executionRunId: f.runId, checkoutRunId: f.runId });
});
it("retains the hold until every exact remote lease is confirmed stopped, then permits ordinary checkout", async () => {
const f = await manuallyHeldRemoteRun();
const scope = { companyId: f.companyId, runId: f.runId, issueId: f.issueId };
expect(await nativeManualAdmissionHoldsExecution(f.db, scope)).toBe(true);
expect(await nativeManualAdmissionHoldsExecution(f.db, { ...scope, companyId: randomUUID() })).toBe(false);
expect(await nativeManualAdmissionHoldsExecution(f.db, { ...scope, runId: randomUUID() })).toBe(false);
expect(await nativeManualAdmissionHoldsExecution(f.db, { ...scope, issueId: randomUUID() })).toBe(false);
const leases = await f.db.insert(environmentLeases).values([1, 2].map(index => ({ companyId: f.companyId,
issueId: f.issueId, heartbeatRunId: f.runId, provider: "daytona", providerLeaseId: `fixture-${index}-${randomUUID()}`,
status: "released", releasedAt: new Date(), cleanupStatus: "success", metadata: {},
}))).returning();
expect(await nativeManualAdmissionHoldsExecution(f.db, scope)).toBe(true);
for (let index = 0; index < leases.length; index += 1) {
const lease = leases[index];
await f.db.update(environmentLeases).set({ metadata: { remoteExecutionTermination: remoteTerminationReceipt(lease,
{ providerLeaseId: lease.providerLeaseId, state: "stopped" }) } }).where(eq(environmentLeases.id, lease.id));
expect(await nativeManualAdmissionHoldsExecution(f.db, scope)).toBe(index < leases.length - 1);
}
expect(await releaseSettledNativeRunProcess(f.db, f.runId)).toBe("released");
expect(await hasPendingRunCleanup(f.db, f.companyId, f.runId)).toBe(false);
expect(await recoveryService(f.db, { enqueueWakeup: vi.fn() }).sweepStaleIssueLocks({ companyId: f.companyId, issueId: f.issueId }))
.toMatchObject({ cleared: 1 });
const actorRunId = randomUUID();
await f.db.insert(heartbeatRuns).values({ id: actorRunId, companyId: f.companyId, agentId: f.agentId,
status: "running", invocationSource: "manual", contextSnapshot: { issueId: f.issueId } });
expect(await issueService(f.db).assertCheckoutOwner(f.issueId, f.agentId, actorRunId))
.toMatchObject({ executionRunId: actorRunId, checkoutRunId: actorRunId });
});
it.each(["ordinary terminal", "accepted result"] as const)("does not preserve unrelated %s ownership", async state => {
const f = await manuallyHeldRemoteRun();
if (state === "ordinary terminal") await f.db.update(nativeRunFinalizations).set({ failureDetail: null })
.where(eq(nativeRunFinalizations.runId, f.runId));
else {
const [run] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
const [result] = await f.db.insert(nativeRunResults).values({ companyId: f.companyId, issueId: f.issueId,
runId: f.runId, completionContractId: run.completionContractId!, serverFingerprint: randomUUID(),
canonicalSha256: "a".repeat(64), schemaStatus: "accepted", resultJson: {} }).returning();
await f.db.update(nativeRunFinalizations).set({ resultId: result.id }).where(eq(nativeRunFinalizations.runId, f.runId));
}
expect(await nativeManualAdmissionHoldsExecution(f.db, { companyId: f.companyId, runId: f.runId, issueId: f.issueId })).toBe(false);
expect(await recoveryService(f.db, { enqueueWakeup: vi.fn() }).sweepStaleIssueLocks({ companyId: f.companyId, issueId: f.issueId }))
.toMatchObject({ cleared: 1 });
});
it("admits an explicit correlated Stop for the preserved manual execution", async () => {
const f = await manuallyHeldRemoteRun();
const requestId = randomUUID();
expect(await claimCancellationRequest(f.db, f.runId, f.companyId, requestId, "operator"))
.toMatchObject({ resultJson: { startupCancellation: { cancellationRequestId: requestId, retryCancellation: true } } });
const stop = await cancelNativeSession(f.runId, "Operator Stop", { db: f.db, scope: "run", cancellationRequestId: requestId });
expect(stop).toMatchObject({ dispatched: false, auditId: expect.any(String) });
expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0])
.toMatchObject({ phase: "terminal_failure", failureCode: "native_retry_cancelled" });
expect(await nativeManualAdmissionHoldsExecution(f.db, { companyId: f.companyId, runId: f.runId, issueId: f.issueId })).toBe(true);
expect(await recoveryService(f.db, { enqueueWakeup: vi.fn() }).sweepStaleIssueLocks({ companyId: f.companyId, issueId: f.issueId }))
.toMatchObject({ cleared: 0 });
expect(await cancelNativeSession(f.runId, "Repeat Operator Stop", { db: f.db, scope: "run", cancellationRequestId: requestId }))
.toMatchObject({ auditId: stop.auditId });
await f.db.update(nativeRunFinalizations).set({ failureCode: "native_restart_recovery_blocked" })
.where(eq(nativeRunFinalizations.runId, f.runId));
expect(await cancelNativeSession(f.runId, "Resume recorded Stop", { db: f.db, scope: "run", cancellationRequestId: requestId }))
.toMatchObject({ auditId: stop.auditId });
expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0])
.toMatchObject({ failureCode: "native_retry_cancelled" });
expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0])
.toMatchObject({ executionRunId: f.runId, checkoutRunId: f.runId });
});
it.each(["classic", "remote", "plugin"] as const)("reclaims the exact failed %s execution before remote inspection", async storage => {
const fixture = await failedRemoteRun(storage);
expect(await claimNativeSessionResumptions({ db: fixture.db, runnerInstanceId: "ordinary-resumer",
runIds: [fixture.runId] })).toEqual([]);
const [claim] = await claimNativeRestartRecoveries({ db: fixture.db,
controller: environment.current.successor, restartKind: "hard", runIds: [fixture.runId] });
expect(claim).toMatchObject({ kind: "reattach_remote_runner", runId: fixture.runId, providerAttempt: 2 });
expect((await fixture.db.select().from(issues).where(eq(issues.id, fixture.issueId)))[0])
.toMatchObject({ executionRunId: fixture.runId, status: "in_progress", assigneeAgentId: fixture.agentId });
expect((await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)))[0])
.toMatchObject({ status: "running", processPid: 1386,
runnerProfileJson: { sessionCheckpoint: fixture.checkpoint } });
expect((await fixture.db.select().from(nativeRunFinalizations)
.where(eq(nativeRunFinalizations.runId, fixture.runId)))[0])
.toMatchObject({ attempt: 2, phase: "observed", recoveryState: "awaiting_runner_reattach" });
});
it.each(["company", "run", "issue"] as const)("does not use a plugin lease from another %s", async scope => {
const f = await failedRemoteRun("plugin");
if (!f.pluginLeaseId) throw new Error("Expected a plugin allocation");
const foreignId = randomUUID();
if (scope === "company") {
await f.db.insert(companies).values({ id: foreignId, name: "Other company", issuePrefix: foreignId.slice(0, 8) });
await f.db.update(environmentLeases).set({ companyId: foreignId }).where(eq(environmentLeases.id, f.pluginLeaseId));
} else if (scope === "run") {
await f.db.insert(heartbeatRuns).values({ id: foreignId, companyId: f.companyId, agentId: f.agentId, status: "failed" });
await f.db.update(environmentLeases).set({ heartbeatRunId: foreignId }).where(eq(environmentLeases.id, f.pluginLeaseId));
} else {
await f.db.insert(issues).values({ id: foreignId, companyId: f.companyId, title: "Other task" });
await f.db.update(environmentLeases).set({ issueId: foreignId }).where(eq(environmentLeases.id, f.pluginLeaseId));
}
const onClaim = vi.fn();
expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor,
restartKind: "hard", runIds: [f.runId], onClaim }))
.toMatchObject([{ kind: "blocked", runId: f.runId, reason: "issue_execution_lock_changed" }]);
expect(onClaim).not.toHaveBeenCalled();
expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0])
.toMatchObject({ executionRunId: null, checkoutRunId: null });
expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0])
.toMatchObject({ phase: "retryable_failure", attempt: 1 });
});
it.each((["remote", "plugin"] as const).flatMap(storage => (["done", "cancelled", "blocked", "paused-agent", "pending-agent-approval", "reassigned",
"dependency", "tree-pause", "native-stop", "operator-stop", "execution-owner", "checkout-owner",
"queued-input", "newer-completed-input", "wrong-checkpoint", "missing-authority"] as const).map(gate => [storage, gate] as const)))
("preserves a %s retry held by %s", async (storage, gate) => {
const f = await failedRemoteRun(storage);
const { db } = f;
if (["done", "cancelled", "blocked"].includes(gate)) {
await db.update(issues).set({ status: gate }).where(eq(issues.id, f.issueId));
} else if (gate === "paused-agent" || gate === "pending-agent-approval") {
await db.update(agents).set({ status: gate === "paused-agent" ? "paused" : "pending_approval" }).where(eq(agents.id, f.agentId));
} else if (gate === "reassigned") {
const nextAgent = randomUUID();
await db.insert(agents).values({ id: nextAgent, companyId: f.companyId, name: "Next owner", status: "active" });
await db.update(issues).set({ assigneeAgentId: nextAgent }).where(eq(issues.id, f.issueId));
} else if (gate === "dependency") {
const dependency = randomUUID();
await db.insert(issues).values({ id: dependency, companyId: f.companyId, title: "Required input", status: "todo" });
await db.insert(issueRelations).values({ companyId: f.companyId, issueId: dependency, relatedIssueId: f.issueId, type: "blocks" });
} else if (gate === "tree-pause") {
const holdId = randomUUID();
await db.insert(issueTreeHolds).values({ id: holdId, companyId: f.companyId, rootIssueId: f.issueId, mode: "pause" });
await db.insert(issueTreeHoldMembers).values({ holdId, companyId: f.companyId, issueId: f.issueId,
issueTitle: "Paused work", issueStatus: "in_progress" });
} else if (gate === "native-stop" || gate === "operator-stop") {
await db.update(heartbeatRuns).set({ resultJson: gate === "native-stop"
? { nativeCancellation: { scope: "run", dispatchState: "pending" } }
: { cancellation: { source: "operator", expected: true, initiator: { type: "user" },
reason: "Stop", recordedAt: new Date().toISOString() } } }).where(eq(heartbeatRuns.id, f.runId));
} else if (["execution-owner", "checkout-owner", "queued-input", "newer-completed-input"].includes(gate)) {
const nextRun = randomUUID();
await db.insert(heartbeatRuns).values({ id: nextRun, companyId: f.companyId, agentId: f.agentId,
issueId: f.issueId, contextSnapshot: { issueId: f.issueId, wakeReason: "issue_comment" },
invocationSource: "manual", createdAt: new Date(Date.now() + 1_000),
status: gate === "queued-input" ? "queued" : gate === "newer-completed-input" ? "succeeded" : "failed" });
if (gate === "execution-owner" || gate === "checkout-owner") await db.update(issues)
.set(gate === "execution-owner" ? { executionRunId: nextRun } : { checkoutRunId: nextRun }).where(eq(issues.id, f.issueId));
} else if (gate === "wrong-checkpoint") {
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
await db.update(heartbeatRuns).set({ runnerProfileJson: { ...run.runnerProfileJson,
sessionCheckpoint: { ...f.checkpoint, identity: { ...f.checkpoint.identity, runId: randomUUID() } },
} }).where(eq(heartbeatRuns.id, f.runId));
} else if (gate === "missing-authority") {
await db.delete(nativeRunnerCheckpoints).where(and(eq(nativeRunnerCheckpoints.companyId, f.companyId),
eq(nativeRunnerCheckpoints.runId, f.runId)));
}
const beforeTask = (await db.select().from(issues).where(eq(issues.id, f.issueId)))[0];
const beforeRun = (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0];
const beforeOwner = (await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0];
const dispositions = await claimNativeRestartRecoveries({ db,
controller: environment.current.successor, restartKind: "hard", runIds: [f.runId] });
expect(dispositions.every(value => value.kind === "blocked")).toBe(true);
expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toEqual(beforeTask);
const [afterRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
const [afterOwner] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
if (["done", "cancelled", "newer-completed-input", "execution-owner", "checkout-owner", "queued-input"].includes(gate)) {
expect(afterOwner).toMatchObject({ phase: "terminal_failure", attempt: 1, nextAttemptAt: null,
failureDetail: { systemRecovery: { state: "exhausted" } } });
expect(afterRun).toMatchObject({ status: "failed", processPid: beforeRun.processPid,
runnerProfileJson: beforeRun.runnerProfileJson });
return;
}
expect(afterRun).toEqual(beforeRun);
expect(afterOwner.failureDetail).toMatchObject(beforeOwner.failureDetail!);
expect({ ...afterOwner, nextAttemptAt: beforeOwner.nextAttemptAt, updatedAt: beforeOwner.updatedAt,
recoveryHistory: beforeOwner.recoveryHistory, failureDetail: beforeOwner.failureDetail }).toEqual(beforeOwner);
if (gate === "execution-owner") expect(afterOwner).toEqual(beforeOwner);
else expect(afterOwner.nextAttemptAt!.getTime()).toBeGreaterThan(beforeOwner.nextAttemptAt!.getTime());
});
it.each(["remote", "plugin"] as const)("waits for the exact old %s controller lease and keeps future retries quiet", async storage => {
const f = await failedRemoteRun(storage);
const now = new Date(), deadline = new Date(now.getTime() + 60_000);
await f.db.update(nativeRunFinalizations).set({ nextAttemptAt: deadline })
.where(eq(nativeRunFinalizations.runId, f.runId));
expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor,
restartKind: "hard", now, runIds: [f.runId] })).toEqual([]);
await f.db.update(nativeRunFinalizations).set({ nextAttemptAt: new Date(0), leaseOwner: "old-owner", leaseExpiresAt: deadline })
.where(eq(nativeRunFinalizations.runId, f.runId));
expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor,
restartKind: "hard", now, runIds: [f.runId] })).toMatchObject([{ kind: "awaiting_evidence", reason: "remote_controller_lease_live" }]);
expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0].executionRunId).toBeNull();
expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0].status).toBe("failed");
});
it.each(["remote", "plugin"] as const)("grants two competing controllers only one original %s execution", async storage => {
const f = await failedRemoteRun(storage);
const claims = (await Promise.all([environment.current.successor,
{ ...environment.current.successor, bootId: randomUUID() }].map(controller =>
claimNativeRestartRecoveries({ db: f.db, controller, restartKind: "hard", runIds: [f.runId] })))).flat();
expect(claims.filter(claim => claim.kind === "reattach_remote_runner")).toHaveLength(1);
expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0])
.toMatchObject({ controllerGeneration: 2, attempt: 2 });
expect(await f.db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, f.runId),
eq(heartbeatRunEvents.eventType, "native.recovery.execution_reclaimed")))).toHaveLength(1);
});
it.each(["remote", "plugin"] as const)("rolls back the %s task, owner, run, and recovery event together", async storage => {
const f = await failedRemoteRun(storage);
await expect(f.db.transaction(async tx => {
const claims = await claimNativeRestartRecoveries({ db: tx as unknown as typeof f.db,
controller: environment.current.successor, restartKind: "hard", runIds: [f.runId] });
expect(claims).toMatchObject([{ kind: "reattach_remote_runner" }]);
throw new Error("rollback admission");
})).rejects.toThrow("rollback admission");
expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0].executionRunId).toBeNull();
expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0].status).toBe("failed");
expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0])
.toMatchObject({ phase: "retryable_failure", controllerGeneration: 1, leaseOwner: null, attempt: 1 });
expect(await f.db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, f.runId),
eq(heartbeatRunEvents.eventType, "native.recovery.execution_reclaimed")))).toEqual([]);
});
it.each([
["succeeded", "2026-11-01T12:00:00.000600Z", 1, true],
["succeeded", "2026-11-01T12:00:00.000000Z", 1, true],
["failed", "2026-11-01T12:00:00.000000Z", 1, true],
["failed", "2026-11-01T12:00:00.000600Z", 1, true],
["succeeded", "2026-11-01T12:00:00.000599Z", 1, false],
["failed", "2026-11-01T12:00:00.000601Z", -1, true],
["failed", "2026-11-01T12:00:00.000600Z", -1, false],
] as const)("uses canonical source order for %s input at %s with ID offset %s", async (status, timestamp, offset, superseded) => {
const f = await failedRemoteRun("remote");
const idHex = (BigInt(`0x${f.runId.replaceAll("-", "")}`) + BigInt(offset)).toString(16).padStart(32, "0");
const nextId = `${idHex.slice(0, 8)}-${idHex.slice(8, 12)}-${idHex.slice(12, 16)}-${idHex.slice(16, 20)}-${idHex.slice(20)}`;
const sourceTimestamp = timestamp.endsWith(".000000Z") ? timestamp : "2026-11-01T12:00:00.000600Z";
await f.db.update(heartbeatRuns).set({ createdAt: sql`${sourceTimestamp}::timestamptz` })
.where(eq(heartbeatRuns.id, f.runId));
await f.db.insert(heartbeatRuns).values({ id: nextId, companyId: f.companyId, agentId: f.agentId,
issueId: f.issueId, invocationSource: "manual", status, createdAt: sql`${timestamp}::timestamptz` });
const [disposition] = await claimNativeRestartRecoveries({ db: f.db,
controller: environment.current.successor, restartKind: "hard", runIds: [f.runId] });
expect(disposition).toMatchObject(superseded
? { kind: "blocked", reason: "native_retry_source_superseded" }
: { kind: "reattach_remote_runner", providerAttempt: 2 });
expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0].executionRunId)
.toBe(superseded ? null : f.runId);
});
it.each((["remote", "plugin"] as const).flatMap(storage => (["released", "retained"] as const).map(lock => [storage, lock] as const)))
("exhausts the existing attempt budget after repeated %s snapshot failures with a %s task lock", async (storage, lock) => {
const f = await failedRemoteRun(storage);
if (lock === "retained") await f.db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId));
const [issue] = await f.db.select().from(issues).where(eq(issues.id, f.issueId));
const [contract] = await f.db.select().from(completionContracts).where(eq(completionContracts.issueId, f.issueId));
const execution: NativeExecutionInputV1 = {
schema: "paperclip.native-execution-input.v1",
binding: { companyId: f.companyId, runId: f.runId, issueId: f.issueId, agentId: f.agentId, executionWorkspaceId: f.runId },
provider: { kind: "codex", model: "gpt-5.6-luna" },
task: { identifier: issue.identifier!, title: issue.title, description: null, prompt: "Continue retained work.", workMode: "standard" },
workspace: { cwd: environment.current.runtimeRoot, repoUrl: null, repoRef: null, branchName: null },
session: { normalizedSessionId: f.native.normalizedSessionId, driverKind: "codex_app_server", protocolVersion: 1,
lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } },
completionContract: { id: contract.id, sha256: contract.canonicalSha256, schemaVersion: "paperclip.completion-contract.v1",
contract: buildNativeCompletionContract(issue, { revision: contract.revision }) },
interactionResponses: [], credentialBindings: [],
};
await f.db.update(heartbeatRuns).set({ runnerProfileJson: sql`coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb)
|| ${JSON.stringify({ nativeExecutionInput: execution })}::jsonb` }).where(eq(heartbeatRuns.id, f.runId));
provider.execute.mockReset().mockRejectedValue(new Error("PRP command session.snapshot failed: original provider unavailable"));
const controller = await currentNativeControllerIdentity();
for (const attempt of [2, 3]) {
await f.db.update(nativeRunFinalizations).set({ nextAttemptAt: new Date(0) }).where(eq(nativeRunFinalizations.runId, f.runId));
const [claim] = await claimNativeRestartRecoveries({ db: f.db, controller, restartKind: "hard", runIds: [f.runId] });
expect(claim).toMatchObject({ kind: "reattach_remote_runner", providerAttempt: attempt });
if (claim.kind !== "reattach_remote_runner") throw new Error("Expected exact remote recovery claim");
await expect(executePaperclipNativeSession({ db: f.db, execution, runnerInstanceId: f.native.runnerInstanceId,
leaseOwner: claim.leaseOwner, restartRecovery: claim,
backend: { descriptor: async () => { throw new Error("Unexpected descriptor call"); },
openSession: async () => { throw new Error("Unexpected session launch"); } },
})).rejects.toThrow("PRP command session.snapshot failed");
const [owner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
expect(owner).toMatchObject({ attempt, phase: attempt === 3 ? "terminal_failure" : "retryable_failure",
failureCode: attempt === 3 ? "native_session_retry_exhausted" : "native_session_interrupted",
recoveryState: attempt === 3 ? "blocked" : "resuming_session", leaseOwner: null });
expect(owner.nextAttemptAt === null).toBe(attempt === 3);
if (lock === "released") await f.db.update(issues).set({ executionRunId: null }).where(eq(issues.id, f.issueId));
}
expect(provider.execute).toHaveBeenCalledTimes(2);
expect(await claimNativeRestartRecoveries({ db: f.db, controller, restartKind: "hard", runIds: [f.runId] })).toEqual([]);
expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toMatchObject({ status: "blocked", executionRunId: null });
expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0])
.toMatchObject({ status: "failed", nativePhase: "terminal_failure", errorCode: "native_session_interrupted",
processPid: 1386, runnerProfileJson: { sessionCheckpoint: f.checkpoint } });
});
it.each((["remote", "plugin"] as const).flatMap(storage => (["released", "retained"] as const).map(lock => [storage, lock] as const)))
("does not admit another failed %s retry when the persisted provider budget is already exhausted with a %s task lock", async (storage, lock) => {
const f = await failedRemoteRun(storage);
if (lock === "retained") await f.db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId));
await f.db.update(nativeRunFinalizations).set({ attempt: 3 }).where(eq(nativeRunFinalizations.runId, f.runId));
expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor,
restartKind: "hard", runIds: [f.runId] })).toMatchObject([{ kind: "blocked", reason: "execution_recovery_budget_exhausted" }]);
expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0].executionRunId).toBeNull();
expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0])
.toMatchObject({ phase: "terminal_failure", attempt: 3, leaseOwner: null, nextAttemptAt: null,
failureCode: "native_restart_recovery_blocked", failureDetail: { reason: "execution_recovery_budget_exhausted" } });
expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0])
.toMatchObject({ status: "failed", nativePhase: "terminal_failure", processPid: 1386,
runnerProfileJson: { sessionCheckpoint: f.checkpoint } });
expect(await f.db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.sourceIssueId, f.issueId),
eq(issueRecoveryActions.kind, "active_run_watchdog"))))
.toMatchObject([{ ownerType: "board", evidence: { runId: f.runId, reason: "execution_recovery_budget_exhausted" } }]);
expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor,
restartKind: "hard", runIds: [f.runId] })).toEqual([]);
});
it("persists a blocked retry deadline and records only changed hold reasons", async () => {
const f = await failedRemoteRun("remote");
await f.db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agentId));
const now = new Date();
const attempt = (at: Date) => claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor,
restartKind: "hard", runIds: [f.runId], now: at });
const [first] = await attempt(now);
expect(first).toMatchObject({ kind: "blocked", reason: "native_retry_admission_ineligible", transitionUnchanged: false });
if (!("retryAt" in first) || !first.retryAt) throw new Error("missing first retry deadline");
expect(first.retryAt.getTime() - now.getTime()).toBeGreaterThanOrEqual(24_000);
expect(first.retryAt.getTime() - now.getTime()).toBeLessThanOrEqual(36_000);
expect(await attempt(new Date(first.retryAt.getTime() - 1))).toEqual([]);
const [second] = await attempt(first.retryAt);
expect(second).toMatchObject({ kind: "blocked", transitionUnchanged: true });
if (!("retryAt" in second) || !second.retryAt) throw new Error("missing second retry deadline");
expect(second.retryAt.getTime() - first.retryAt.getTime()).toBeGreaterThanOrEqual(48_000);
let [owner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
expect(owner).toMatchObject({ phase: "retryable_failure", attempt: 1,
nextAttemptAt: second.retryAt, recoveryHistory: [{ reason: "native_retry_admission_ineligible" }] });
await f.db.update(agents).set({ status: "active" }).where(eq(agents.id, f.agentId));
await f.db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, issueId: f.issueId,
status: "succeeded", createdAt: new Date(now.getTime() + 1_000) });
expect(await attempt(second.retryAt)).toMatchObject([{ kind: "blocked", reason: "native_retry_source_superseded" }]);
[owner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
expect(owner.recoveryHistory).toMatchObject([{ reason: "native_retry_admission_ineligible" }, { reason: "native_retry_source_superseded" }]);
});
it("registers recovery work intent before writing a blocked retry deadline", async () => {
const f = await failedRemoteRun("remote");
await f.db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agentId));
const transaction = f.db.transaction.bind(f.db);
let writes = 0;
const interception = vi.spyOn(f.db, "transaction").mockImplementation((async (callback, options) => transaction(async tx => {
const update = tx.update.bind(tx);
const updates = vi.spyOn(tx, "update").mockImplementation((table => {
if (Object.is(table, nativeRunFinalizations)) {
writes += 1;
expect(databaseWorkPending(f.db, DELIVERY_QUEUES.nativeProcessSessions)).toBe(true);
}
return update(table);
}) as typeof tx.update);
try { return await callback(tx); } finally { updates.mockRestore(); }
}, options)) as typeof f.db.transaction);
try {
expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor,
restartKind: "hard", runIds: [f.runId] })).toMatchObject([{ kind: "blocked" }]);
expect(writes).toBe(1);
} finally { interception.mockRestore(); }
});
it.each(["done", "cancelled", "superseded"] as const)("settles a permanently %s remote retry without recurring work", async gate => {
for (const lock of ["released", "retained"]) {
const f = await failedRemoteRun("remote");
if (lock === "retained") await f.db.update(issues).set({ executionRunId: f.runId, checkoutRunId: f.runId })
.where(eq(issues.id, f.issueId));
if (gate === "superseded") await f.db.insert(heartbeatRuns).values({ companyId: f.companyId,
agentId: f.agentId, issueId: f.issueId, status: "succeeded", createdAt: new Date(Date.now() + 1_000) });
else await f.db.update(issues).set({ status: gate }).where(eq(issues.id, f.issueId));
const onClaim = vi.fn();
const environmentWake = vi.fn(), stopWatchingEnvironment = subscribeDeliveryWork(f.db, DELIVERY_QUEUES.nativeEnvironmentSettlement, environmentWake);
const [disposition] = await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor,
restartKind: "hard", runIds: [f.runId], onClaim });
expect(disposition).toMatchObject({ kind: "blocked" });
expect(disposition).not.toHaveProperty("retryAt");
expect(onClaim).not.toHaveBeenCalled();
expect(environmentWake).not.toHaveBeenCalled();
stopWatchingEnvironment();
expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toMatchObject({
executionRunId: lock === "retained" ? f.runId : null, checkoutRunId: lock === "retained" ? f.runId : null });
expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0])
.toMatchObject({ phase: "terminal_failure", nextAttemptAt: null, attempt: 1 });
expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0])
.toMatchObject({ processPid: 1386, runnerProfileJson: { sessionCheckpoint: f.checkpoint } });
const wake = vi.fn(), unsubscribe = subscribeDeliveryWork(f.db, DELIVERY_QUEUES.nativeProcessSessions, wake);
const transaction = vi.spyOn(f.db, "transaction");
try {
for (const elapsed of [30_000, 60_000, 3_600_000]) expect(await claimNativeRestartRecoveries({ db: f.db,
controller: environment.current.successor, restartKind: "hard", runIds: [f.runId],
now: new Date(Date.now() + elapsed), onClaim })).toEqual([]);
expect(transaction).not.toHaveBeenCalled();
expect(wake).not.toHaveBeenCalled();
} finally { transaction.mockRestore(); unsubscribe(); }
}
});
it.each(["paused", "Stop", "missing-authority"] as const)("bounds %s holds and preserves the provider and own task lock for manual reconciliation", async gate => {
const f = await failedRemoteRun("remote");
if (gate === "paused") await f.db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agentId));
if (gate === "Stop") await f.db.update(heartbeatRuns).set({ resultJson: {
nativeCancellation: { scope: "run", dispatchState: "pending" },
} }).where(eq(heartbeatRuns.id, f.runId));
if (gate === "missing-authority") await f.db.delete(nativeRunnerCheckpoints)
.where(and(eq(nativeRunnerCheckpoints.companyId, f.companyId), eq(nativeRunnerCheckpoints.runId, f.runId)));
await f.db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId));
await f.db.update(nativeRunFinalizations).set({ attempt: 3 }).where(eq(nativeRunFinalizations.runId, f.runId));
let now = new Date();
const onClaim = vi.fn();
const environmentWake = vi.fn(), stopWatchingEnvironment = subscribeDeliveryWork(f.db, DELIVERY_QUEUES.nativeEnvironmentSettlement, environmentWake);
for (let index = 0; index <= SYSTEM_RECOVERY_MAX_RETRIES; index += 1) {
const [disposition] = await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor,
restartKind: "hard", runIds: [f.runId], now, onClaim });
const [owner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
const policy = readSystemRecovery(owner.failureDetail);
expect(policy).not.toBeNull();
expect(owner.attempt).toBe(3);
if (index < SYSTEM_RECOVERY_MAX_RETRIES) {
expect(owner.phase).toBe("retryable_failure");
expect(policy).toMatchObject({ state: "awaiting_evidence" });
expect(disposition).toHaveProperty("retryAt", owner.nextAttemptAt);
now = owner.nextAttemptAt!;
} else {
expect(owner).toMatchObject({ phase: "terminal_failure", nextAttemptAt: null });
expect(policy).toMatchObject({ state: "exhausted" });
expect(disposition).not.toHaveProperty("retryAt");
}
}
expect(onClaim).not.toHaveBeenCalled();
expect(environmentWake).not.toHaveBeenCalled();
stopWatchingEnvironment();
expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toMatchObject({ executionRunId: f.runId });
expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0])
.toMatchObject({ processPid: 1386, runnerProfileJson: { sessionCheckpoint: f.checkpoint } });
expect(await f.db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)))
.toMatchObject([{ ownerType: "board", wakePolicy: null,
evidence: { automaticAdmissionExhausted: true, executorStopEvidence: "unverified", providerPid: 1386, checkpointPreserved: true },
nextAction: expect.stringContaining("original executor stop and state remain unverified") }]);
expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor,
restartKind: "hard", runIds: [f.runId], now: new Date(now.getTime() + 3_600_000) })).toEqual([]);
});
it("settles the old admission policy while preserving a newer active execution and its binding", async () => {
const f = await failedRemoteRun("remote");
const newerRunId = randomUUID();
await f.db.insert(heartbeatRuns).values({ id: newerRunId, companyId: f.companyId, agentId: f.agentId,
issueId: f.issueId, status: "running", invocationSource: "manual", createdAt: new Date(Date.now() + 1_000) });
await f.db.update(issues).set({ executionRunId: newerRunId, checkoutRunId: newerRunId }).where(eq(issues.id, f.issueId));
const [task] = await f.db.select().from(issues).where(eq(issues.id, f.issueId));
const [newer] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, newerRunId));
const environmentWake = vi.fn(), unsubscribe = subscribeDeliveryWork(f.db, DELIVERY_QUEUES.nativeEnvironmentSettlement, environmentWake);
const onClaim = vi.fn();
try {
expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor,
restartKind: "hard", runIds: [f.runId], onClaim }))
.toMatchObject([{ kind: "blocked", reason: "native_retry_source_superseded" }]);
expect(onClaim).not.toHaveBeenCalled();
expect(environmentWake).not.toHaveBeenCalled();
expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toEqual(task);
expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, newerRunId)))[0]).toEqual(newer);
expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0])
.toMatchObject({ phase: "terminal_failure", nextAttemptAt: null, failureDetail: { systemRecovery: { state: "exhausted" } } });
expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0])
.toMatchObject({ processPid: 1386, runnerProfileJson: { sessionCheckpoint: f.checkpoint } });
const transactions = vi.spyOn(f.db, "transaction");
try {
expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor,
restartKind: "hard", runIds: [f.runId], now: new Date(Date.now() + 3_600_000) })).toEqual([]);
expect(transactions).not.toHaveBeenCalled();
} finally { transactions.mockRestore(); }
} finally { unsubscribe(); }
});
it("resumes the original provider when eligibility returns before admission backoff exhausts", async () => {
const f = await failedRemoteRun("remote");
await f.db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agentId));
await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor,
restartKind: "hard", runIds: [f.runId] });
const [owner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
await f.db.update(agents).set({ status: "active" }).where(eq(agents.id, f.agentId));
expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor,
restartKind: "hard", runIds: [f.runId], now: owner.nextAttemptAt! }))
.toMatchObject([{ kind: "reattach_remote_runner", providerAttempt: 2, runId: f.runId }]);
expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0])
.toMatchObject({ status: "running", processPid: 1386, runnerProfileJson: { sessionCheckpoint: f.checkpoint } });
expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0].executionRunId).toBe(f.runId);
});
it("does not settle permanently ineligible work while the original controller is active", async () => {
const f = await failedRemoteRun("remote");
await f.db.update(issues).set({ status: "done" }).where(eq(issues.id, f.issueId));
const [before] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
const wake = vi.fn(), unsubscribe = subscribeDeliveryWork(f.db, DELIVERY_QUEUES.nativeProcessSessions, wake);
try {
expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor,
restartKind: "hard", runIds: [f.runId], liveRunExecutions: new Set([f.runId]) }))
.toMatchObject([{ kind: "awaiting_evidence", reason: "current_controller_preparing" }]);
expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toEqual(before);
expect(wake).not.toHaveBeenCalled();
expect(await f.db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId))).toEqual([]);
} finally { unsubscribe(); }
});
it("keeps a retry deadline that changed after discovery read-only and quiet", async () => {
const f = await failedRemoteRun("remote");
await f.db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agentId));
const deadline = new Date(Date.now() + 60_000);
const transaction = f.db.transaction.bind(f.db);
const interception = vi.spyOn(f.db, "transaction").mockImplementationOnce((async (callback, options) => {
await f.db.update(nativeRunFinalizations).set({ nextAttemptAt: deadline }).where(eq(nativeRunFinalizations.runId, f.runId));
return transaction(callback, options);
}) as typeof f.db.transaction);
const wake = vi.fn(), unsubscribe = subscribeDeliveryWork(f.db, DELIVERY_QUEUES.nativeProcessSessions, wake);
try {
expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor,
restartKind: "hard", runIds: [f.runId] })).toMatchObject([{ kind: "blocked", retryAt: deadline, transitionUnchanged: true }]);
const [owner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
expect(owner).toMatchObject({ nextAttemptAt: deadline, recoveryHistory: [], phase: "retryable_failure", attempt: 1 });
expect(wake).not.toHaveBeenCalled();
} finally { interception.mockRestore(); unsubscribe(); }
});
it("rolls back work intent when the locked retry deadline update matches no row", async () => {
const f = await failedRemoteRun("remote");
await f.db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agentId));
const name = `retry_noop_${f.runId.replaceAll("-", "")}`;
await f.db.execute(sql.raw(`create function ${name}() returns trigger language plpgsql as $$begin return null; end$$`));
await f.db.execute(sql.raw(`create trigger ${name} before update on native_run_finalizations for each row
when (old.run_id = '${f.runId}'::uuid) execute function ${name}()`));
const [before] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
try {
const now = new Date();
await expect(claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor,
restartKind: "hard", runIds: [f.runId], now })).resolves.toEqual([{ kind: "awaiting_evidence",
runId: f.runId, reason: NATIVE_RESTART_RECOVERY_FAILED, retryAt: new Date(now.getTime() + 30_000) }]);
expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toEqual(before);
await reconcileDatabaseWork(f.db, DELIVERY_QUEUES.nativeProcessSessions);
expect(databaseWorkPending(f.db, DELIVERY_QUEUES.nativeProcessSessions)).toBe(false);
} finally {
await f.db.execute(sql.raw(`drop trigger ${name} on native_run_finalizations`));
await f.db.execute(sql.raw(`drop function ${name}()`));
}
});
it("does not publish or log a deferred transition when an ineligible retry still has a live lease", async () => {
const f = await failedRemoteRun("remote");
const now = new Date(), leaseDeadline = new Date(now.getTime() + 60_000);
await f.db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agentId));
await f.db.update(nativeRunFinalizations).set({ leaseOwner: "live-controller", leaseExpiresAt: leaseDeadline })
.where(eq(nativeRunFinalizations.runId, f.runId));
const [before] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
const wake = vi.fn(), unsubscribe = subscribeDeliveryWork(f.db, DELIVERY_QUEUES.nativeProcessSessions, wake);
try {
for (let index = 0; index < 2; index += 1) {
expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor,
restartKind: "hard", runIds: [f.runId], now })).toMatchObject([{ kind: "blocked",
retryAt: leaseDeadline, transitionUnchanged: true }]);
expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toEqual(before);
expect(wake).not.toHaveBeenCalled();
}
} finally { unsubscribe(); }
});
it.each(["released", "retained"] as const)("keeps exhausted %s retries fenced behind live owners, Stops and newer input", async lock => {
for (const gate of ["live-owner", "Stop", "newer-input", "missing-authority"]) {
const f = await failedRemoteRun("remote");
if (lock === "retained") await f.db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId));
await f.db.update(nativeRunFinalizations).set({ attempt: 3,
...(gate === "live-owner" ? { leaseOwner: "live-controller", leaseExpiresAt: new Date(Date.now() + 60_000) } : {}),
}).where(eq(nativeRunFinalizations.runId, f.runId));
if (gate === "Stop") await f.db.update(heartbeatRuns).set({ resultJson: {
nativeCancellation: { scope: "run", dispatchState: "pending" },
} }).where(eq(heartbeatRuns.id, f.runId));
if (gate === "newer-input") await f.db.insert(heartbeatRuns).values({ companyId: f.companyId,
agentId: f.agentId, issueId: f.issueId, status: "succeeded", createdAt: new Date(Date.now() + 1_000) });
if (gate === "missing-authority") await f.db.delete(nativeRunnerCheckpoints)
.where(and(eq(nativeRunnerCheckpoints.companyId, f.companyId), eq(nativeRunnerCheckpoints.runId, f.runId)));
const onClaim = vi.fn();
expect((await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor,
restartKind: "hard", runIds: [f.runId], onClaim }))[0].kind).toBe(gate === "live-owner" ? "awaiting_evidence" : "blocked");
expect(onClaim).not.toHaveBeenCalled();
expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0].executionRunId).toBe(lock === "retained" ? f.runId : null);
expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0])
.toMatchObject({ phase: gate === "newer-input" ? "terminal_failure" : "retryable_failure", attempt: 3 });
const actions = await f.db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
if (gate === "newer-input") expect(actions).toMatchObject([{ ownerType: "board", wakePolicy: null }]);
else expect(actions).toEqual([]);
}
});
it("preserves the charged attempt when replacing an active remote controller", async () => {
const f = await failedRemoteRun("remote");
await f.db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId));
await f.db.update(heartbeatRuns).set({ status: "running", nativePhase: "observed" }).where(eq(heartbeatRuns.id, f.runId));
await f.db.update(nativeRunFinalizations).set({ attempt: 2, phase: "observed", failureCode: null, failureDetail: null })
.where(eq(nativeRunFinalizations.runId, f.runId));
expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor,
restartKind: "hard", runIds: [f.runId] })).toMatchObject([{ kind: "reattach_remote_runner", providerAttempt: 2 }]);
expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0])
.toMatchObject({ phase: "observed", attempt: 2, controllerGeneration: 2 });
});
it("ignores another company's newer input when reclaiming this company's execution", async () => {
const f = await failedRemoteRun("remote");
const foreignCompany = randomUUID(), foreignAgent = randomUUID();
await f.db.insert(companies).values({ id: foreignCompany, name: "Other company", issuePrefix: "FOREIGN" });
await f.db.insert(agents).values({ id: foreignAgent, companyId: foreignCompany, name: "Other agent", status: "active" });
await f.db.insert(heartbeatRuns).values({ companyId: foreignCompany, agentId: foreignAgent,
status: "queued", contextSnapshot: { issueId: f.issueId }, createdAt: new Date(Date.now() + 1_000) });
expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor,
restartKind: "hard", runIds: [f.runId] })).toMatchObject([{ kind: "reattach_remote_runner" }]);
});
});
server/src/services/native-runtime/native-restart-recovery-isolation.test.ts
+32 -3
View File
@@ -1,6 +1,9 @@
import { randomUUID } from "node:crypto";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { resolve } from "node:path";
import { and, eq, inArray, sql } from "drizzle-orm";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
import {
agents, applyPendingMigrations, companies, createDb, environmentLeases, environments, heartbeatRunEvents,
heartbeatRuns, issues, nativeRunFinalizations,
@@ -9,6 +12,8 @@ import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
import { heartbeatService } from "../heartbeat.js";
import { claimNativeRestartRecoveries, NATIVE_RESTART_RECOVERY_FAILED } from "./native-restart-recovery.js";
import { DurablePrpControlPlane } from "../../vendor/paperclip-runner/index.js";
import { nativeRunnerCheckpointWriter } from "./native-runner-checkpoint.js";
const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_TEST_DATABASE_URL");
const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport();
@@ -19,6 +24,7 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
// whose claim fails for an unrelated reason. Neither may stop the batch.
(support.supported ? describe : describe.skip)("native restart recovery isolates one run's failure", () => {
let db: ReturnType<typeof createDb>;
let runtimeRoot: string;
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
const companyId = randomUUID();
const agentId = randomUUID();
@@ -32,8 +38,14 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
}
await db.insert(companies).values({ id: companyId, name: "Restart isolation", issuePrefix: companyId.slice(0, 8) });
await db.insert(agents).values({ id: agentId, companyId, name: "Engineer", adapterType: "paperclip_runner" });
runtimeRoot = await mkdtemp(resolve(tmpdir(), "restart-isolation-authority-"));
vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "83".repeat(32));
}, 30_000);
afterAll(async () => { await db?.$client.end(); await temporary?.cleanup(); });
afterAll(async () => {
await db?.$client.end(); await temporary?.cleanup();
if (runtimeRoot) await rm(runtimeRoot, { recursive: true, force: true });
vi.unstubAllEnvs();
});
async function seedScheduledResume(options: { lock: "self" | "none"; routine?: { id: string; fingerprint: string } }) {
const issueId = randomUUID(), runId = randomUUID(), sessionId = randomUUID();
@@ -51,6 +63,7 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
environmentId: environment.id, provider: "daytona", providerLeaseId: randomUUID(), status: "active",
leasePolicy: "ephemeral", metadata: { driver: "sandbox", remoteCwd: "/work" } }).returning();
await db.update(heartbeatRuns).set({ runnerProfileJson: {
nativeExecutionInput: { binding: { companyId, issueId, runId, agentId } },
nativeWorkspaceSync: {
schema: "paperclip.native-workspace-sync/v3", state: "prepared", descriptorSha256: "a".repeat(64),
baselineSha256: "b".repeat(64), finalHostSha256: null, workspaceId: runId,
@@ -63,9 +76,25 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
await db.update(issues).set({ executionRunId: runId, checkoutRunId: runId, executionLockedAt: failedAt })
.where(eq(issues.id, issueId));
}
const owner = { bootId: randomUUID(), generation: 1, leaseOwner: randomUUID() };
await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "retryable_failure", attempt: 1,
controllerBootId: owner.bootId, controllerGeneration: owner.generation,
leaseOwner: owner.leaseOwner, leaseExpiresAt: new Date(Date.now() + 60_000),
nextAttemptAt: new Date(Date.now() - 60_000), recoveryState: "resuming_session",
failureCode: "native_session_interrupted" });
failureCode: "native_session_interrupted",
failureDetail: { recoveryMode: "exact_checkpoint_resume", checkpointExists: true,
providerSessionEstablished: true, providerEventsExist: false } });
const authority = new DurablePrpControlPlane({ stateDirectory: resolve(runtimeRoot, runId),
identity: { runId, normalizedSessionId: sessionId, runnerInstanceId: randomUUID(),
environmentLeaseId: lease.id, turnId: randomUUID(), itemId: randomUUID() },
expectedRunnerVersion: "restart-isolation-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` });
authority.setCheckpointPersistence(nativeRunnerCheckpointWriter(db, {
companyId, issueId, runId, authorityRunId: runId, ...owner,
}));
await authority.flushCheckpoint();
await authority.stop();
await db.update(nativeRunFinalizations).set({ leaseOwner: null, leaseExpiresAt: null })
.where(eq(nativeRunFinalizations.runId, runId));
return { issueId, runId };
}
server/src/services/native-runtime/native-restart-recovery.ts
+137 -66
View File
@@ -1,4 +1,5 @@
import { readPendingNativeRunStop } from "./native-cancellation-request.js";
import { readNativeSessionResumptionAgent, restoreNativeSessionExecution } from "./native-session-resumption-admission.js";
import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js";
import { NATIVE_RETAINED_OWNER_WAIT } from "./native-controller-recovery.js";
import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js";
@@ -20,7 +21,8 @@ import {
recordNativeLocalProcessStop,
} from "../native-local-process-stop.js";
import { randomUUID } from "node:crypto";
import { and, desc, eq, inArray, isNotNull, isNull, lte, ne, or, sql } from "drizzle-orm";
import { and, desc, eq, gt, inArray, isNotNull, isNull, lte, ne, or, sql } from "drizzle-orm";
import { alias } from "drizzle-orm/pg-core";
import type { Db } from "@paperclipai/db";
import {
agents,
@@ -42,8 +44,9 @@ import { isNativeRunnerOwnershipHeld } from "./native-runner-ownership.js";
import { nativeRestartInterruptedTurnId, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js";
import { retireSupersededNativeRetry } from "./native-superseded-retry.js";
import { legacyWorkspaceRecoveryNotMigrated } from "../legacy-workspace-recovery-migration.js";
import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js";
import { requestNativeStartupCleanup, readNativeStartupContext, promoteNativeStartupCleanup } from "./native-startup-cleanup.js";
import { createSystemRecovery, failSystemRecovery, readSystemRecovery, reserveSystemRecoveryAttempt,
SystemRecoveryFailure, type SystemRecoveryState } from "../system-recovery-state.js";
export type NativeControllerIdentity = {
bootId: string;
@@ -110,6 +113,7 @@ export type NativeRestartRecoveryDisposition =
runId: string;
reason: string;
retryAt?: Date;
transitionUnchanged?: boolean;
};
export function nextNativeProviderAttempt(
@@ -491,11 +495,6 @@ function appendBoundedRecoveryHistory(entry: Record<string, unknown>) {
)`;
}
/**
* Claims abandoned local runner executions without changing the provider retry
* attempt. The transaction locks the run, coordinator, and issue execution
* owner so two successor servers cannot both reconstruct the same authority.
*/
export async function claimNativeRestartRecoveries(input: {
db: Db;
controller?: NativeControllerIdentity;
@@ -521,6 +520,7 @@ export async function claimNativeRestartRecoveries(input: {
.select({
runId: heartbeatRuns.id,
issueId: nativeRunFinalizations.issueId,
companyId: heartbeatRuns.companyId,
})
.from(heartbeatRuns)
.innerJoin(agents, eq(agents.id, heartbeatRuns.agentId))
@@ -579,9 +579,11 @@ export async function claimNativeRestartRecoveries(input: {
// Defer that concrete recovery obligation instead of failing startup
// or waiting while holding a lock the serving controller needs.
const [issue] = await tx
.select({ id: issues.id })
.select({ id: issues.id, companyId: issues.companyId, status: issues.status,
executionRunId: issues.executionRunId, checkoutRunId: issues.checkoutRunId,
assigneeAgentId: issues.assigneeAgentId })
.from(issues)
.where(eq(issues.id, candidate.issueId))
.where(and(eq(issues.id, candidate.issueId), eq(issues.companyId, candidate.companyId)))
.for("update", { skipLocked: true });
if (!issue) return busy();
const row = await tx
@@ -637,20 +639,91 @@ export async function claimNativeRestartRecoveries(input: {
if (!input.liveRunExecutions?.has(row.run.id) && await retireSupersededNativeRetry(tx as unknown as Db, {
run: row.run, coordinator: row.coordinator, issueExecutionRunId: row.issueExecutionRunId, now,
})) return { kind: "blocked", runId: row.run.id, reason: "native_retry_superseded" } as const;
// A scheduled same-run resume can find its task lock released while no
// other run holds the task. Like the local resume claim, reclaim the
// free lock unless another run is admitted for this task.
const reclaimsExecution = row.issueExecutionRunId === null
&& row.coordinator.phase === "retryable_failure"
&& (row.issueCheckoutRunId === null || row.issueCheckoutRunId === row.run.id)
&& !(await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
eq(heartbeatRuns.companyId, row.run.companyId), ne(heartbeatRuns.id, row.run.id),
inArray(heartbeatRuns.status, ["queued", "running", "scheduled_retry"]),
or(eq(heartbeatRuns.issueId, row.coordinator.issueId), eq(heartbeatRuns.nativeIssueId, row.coordinator.issueId),
sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${row.coordinator.issueId}`),
)).limit(1)).length;
if (row.issueExecutionRunId !== row.run.id && !(stopCleanup && row.issueExecutionRunId === null)
&& !reclaimsExecution) {
const profile = row.run.runnerProfileJson ?? {};
const storedRemoteWorkspace = remoteWorkspaceRunReferenceSchema.safeParse(profile.remoteWorkspace);
const remoteWorkspace = readNativeWorkspaceSyncReference(profile.nativeWorkspaceSync)
?? (storedRemoteWorkspace.success ? storedRemoteWorkspace.data : null);
const [pluginTask] = await tx.select({ id: environmentLeases.id,
metadata: sql<Record<string, unknown>>`jsonb_build_object('runnerTaskStartup', ${environmentLeases.metadata}->'runnerTaskStartup')`,
}).from(environmentLeases).where(and(
eq(environmentLeases.companyId, row.run.companyId), eq(environmentLeases.issueId, issue.id),
eq(environmentLeases.heartbeatRunId, row.run.id), sql`${environmentLeases.metadata}->>'runnerTaskLease' = 'true'`,
)).orderBy(desc(environmentLeases.createdAt)).limit(1);
const failedRemoteRetry = row.run.status === "failed" && row.coordinator.phase === "retryable_failure"
&& row.coordinator.failureDetail?.recoveryMode === "exact_checkpoint_resume" && (remoteWorkspace !== null || pluginTask !== undefined);
const releasedRemoteRetry = failedRemoteRetry && row.issueExecutionRunId === null;
const exhaustedRemoteRetry = failedRemoteRetry && row.coordinator.attempt >= 3;
const needsResumptionAdmission = releasedRemoteRetry || exhaustedRemoteRetry ||
(failedRemoteRetry && ["done", "cancelled"].includes(issue.status));
const admission = { exhausted: null as { reason: string; policy: SystemRecoveryState } | null };
const deferRemoteRetry = async (reason: string, permanent = false) => {
if (input.liveRunExecutions?.has(row.run.id)) {
return { kind: "awaiting_evidence", runId: row.run.id, reason: "current_controller_preparing" } as const;
}
const existingDeadline = Math.max(row.coordinator.nextAttemptAt?.getTime() ?? 0,
row.coordinator.leaseOwner ? row.coordinator.leaseExpiresAt?.getTime() ?? 0 : 0);
if (existingDeadline > now.getTime()) {
return { kind: "blocked", runId: row.run.id, reason, retryAt: new Date(existingDeadline),
transitionUnchanged: true } as const;
}
const previousPolicy = readSystemRecovery(row.coordinator.failureDetail);
const policy = previousPolicy?.sourceRunId === row.run.id ? previousPolicy : createSystemRecovery(row.run.id);
const reserved = reserveSystemRecoveryAttempt(policy, now.getTime());
const failed = reserved ? failSystemRecovery(reserved, new SystemRecoveryFailure(
permanent ? "source_binding_invalid" : "allocation_ownership_ambiguous",
permanent ? "user_action_required" : "recheck_evidence"), { now: now.getTime(), evidenceFingerprint: reason })
: { ...policy, state: "exhausted" as const, nextAttemptAt: null };
if (failed.state === "exhausted") {
admission.exhausted = { reason, policy: failed };
return null;
}
const retryAt = new Date(failed.nextAttemptAt!);
const previous = row.coordinator.recoveryHistory.at(-1);
const transitionUnchanged = previous?.disposition === "blocked" && previous.reason === reason;
await notifyLiveRunRecovery(tx, { companyId: row.run.companyId, runId: row.run.id });
const updated = await tx.update(nativeRunFinalizations).set({ nextAttemptAt: retryAt, updatedAt: now,
failureDetail: { ...row.coordinator.failureDetail, systemRecovery: failed },
...(!transitionUnchanged ? { recoveryHistory: appendBoundedRecoveryHistory({
at: now.toISOString(), restartKind: input.restartKind, controllerBootId: controller.bootId,
controllerGeneration: row.coordinator.controllerGeneration, disposition: "blocked", reason,
recoveryRequestId: input.recoveryRequestId ?? null, providerAttempt: row.coordinator.attempt,
stateRootAction: "preserve_fail_closed", retryAt: retryAt.toISOString(),
}) } : {}),
}).where(and(eq(nativeRunFinalizations.companyId, row.run.companyId),
eq(nativeRunFinalizations.issueId, issue.id), eq(nativeRunFinalizations.runId, row.run.id),
eq(nativeRunFinalizations.phase, "retryable_failure"), eq(nativeRunFinalizations.attempt, row.coordinator.attempt),
isNull(nativeRunFinalizations.resultId),
or(isNull(nativeRunFinalizations.nextAttemptAt), lte(nativeRunFinalizations.nextAttemptAt, now)),
or(isNull(nativeRunFinalizations.leaseOwner), isNull(nativeRunFinalizations.leaseExpiresAt), lte(nativeRunFinalizations.leaseExpiresAt, now)),
)).returning({ runId: nativeRunFinalizations.runId });
if (!updated.length) throw new Error("native_retry_deferral_claim_changed");
return { kind: "blocked", runId: row.run.id, reason, retryAt, transitionUnchanged } as const;
};
if (failedRemoteRetry) {
const source = alias(heartbeatRuns, "native_retry_source");
const [newer] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns)
.innerJoin(source, and(eq(source.companyId, row.run.companyId), eq(source.id, row.run.id)))
.where(and(
eq(heartbeatRuns.companyId, row.run.companyId),
or(gt(heartbeatRuns.createdAt, source.createdAt), and(
eq(heartbeatRuns.createdAt, source.createdAt), gt(heartbeatRuns.id, source.id),
)),
or(eq(heartbeatRuns.issueId, issue.id), eq(heartbeatRuns.nativeIssueId, issue.id),
sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issue.id}`),
)).limit(1);
if (newer) {
const deferred = await deferRemoteRetry("native_retry_source_superseded", true);
if (deferred) return deferred;
}
}
const resumptionAgent = needsResumptionAdmission
? await readNativeSessionResumptionAgent(tx as unknown as Db, { issue, run: row.run, coordinator: row.coordinator })
: null;
if (needsResumptionAdmission && !resumptionAgent && !admission.exhausted) {
const deferred = await deferRemoteRetry("native_retry_admission_ineligible", ["done", "cancelled"].includes(issue.status));
if (deferred) return deferred;
}
if (row.issueExecutionRunId !== row.run.id && !(stopCleanup && row.issueExecutionRunId === null) && !releasedRemoteRetry && !admission.exhausted) {
return {
kind: "blocked",
runId: row.run.id,
@@ -676,14 +749,6 @@ export async function claimNativeRestartRecoveries(input: {
return { kind: "awaiting_evidence", runId: row.run.id, reason: "native_startup_cleanup_pending", retryAt: startupRetryAt } as const;
}
const profile = row.run.runnerProfileJson ?? {};
const storedRemoteWorkspace = remoteWorkspaceRunReferenceSchema.safeParse(profile.remoteWorkspace);
const remoteWorkspace = readNativeWorkspaceSyncReference(profile.nativeWorkspaceSync)
?? (storedRemoteWorkspace.success ? storedRemoteWorkspace.data : null);
const [pluginTask] = await tx.select({ id: environmentLeases.id, metadata: environmentLeases.metadata }).from(environmentLeases).where(and(
eq(environmentLeases.companyId, row.run.companyId), eq(environmentLeases.heartbeatRunId, row.run.id),
sql`${environmentLeases.metadata}->>'runnerTaskLease' = 'true'`,
)).orderBy(desc(environmentLeases.createdAt)).limit(1);
const startup = pluginTask?.metadata?.runnerTaskStartup as { contextId?: unknown; deadlineAtMs?: unknown; runId?: unknown } | undefined;
if (startup?.runId === row.run.id && typeof startup.contextId === "string" &&
Number.isSafeInteger(startup.deadlineAtMs) && Number(startup.deadlineAtMs) <= now.getTime() &&
@@ -961,10 +1026,22 @@ export async function claimNativeRestartRecoveries(input: {
const ownershipChanged =
row.issueAssigneeAgentId !== row.run.agentId ||
["done", "cancelled"].includes(row.issueStatus);
const claimKind = ownershipChanged ? null : classification.claimKind;
const reason = ownershipChanged
if (needsResumptionAdmission && (!durableAuthority || ownershipChanged || classification.claimKind !== "reattach_remote_runner")) {
if (!admission.exhausted) {
const deferred = await deferRemoteRetry("native_retry_remote_authority_unverified");
if (deferred) return deferred;
}
}
const hold = admission.exhausted;
const preserveExecutionBinding = hold !== null;
const claimKind = hold || ownershipChanged || exhaustedRemoteRetry ? null : classification.claimKind;
const reason = hold?.reason ?? (ownershipChanged
? "task_ownership_or_status_changed"
: classification.reason;
: exhaustedRemoteRetry ? "execution_recovery_budget_exhausted" : classification.reason);
const nextAction = hold
? "Automatic admission checks exhausted. The original executor stop and state remain unverified. The provider PID, checkpoint and task bindings are preserved; reconcile that execution before starting fresh work."
: "Inspect the preserved checkpoint and reconcile the previous execution before starting a fresh session.";
if (!claimKind) {
const generation = row.coordinator.controllerGeneration;
@@ -999,9 +1076,9 @@ export async function claimNativeRestartRecoveries(input: {
failureCode: "native_restart_recovery_blocked",
failureDetail: {
...row.coordinator.failureDetail,
...(hold ? { systemRecovery: hold.policy } : {}),
reason,
nextAction:
"Inspect the preserved checkpoint and reconcile the previous execution before starting a fresh session.",
nextAction,
},
recoveryState: "blocked",
recoveryRequestId: input.recoveryRequestId ?? null,
@@ -1015,11 +1092,13 @@ export async function claimNativeRestartRecoveries(input: {
eq(nativeRunFinalizations.phase, row.coordinator.phase),
),
);
await notifyExternalAgentSettlement(tx, { companyId: row.run.companyId, runId: row.run.id });
await notifyNativeEnvironmentSettlement(tx, { companyId: row.run.companyId, runId: row.run.id });
await notifyInstructionCollection(tx, { companyId: row.run.companyId, runId: row.run.id });
await notifyCancelledWorkspacePreservation(tx, { companyId: row.run.companyId, runId: row.run.id });
await notifyUnsafeWorkspaceExportRunChange(tx, row.run);
if (!preserveExecutionBinding) {
await notifyExternalAgentSettlement(tx, { companyId: row.run.companyId, runId: row.run.id });
await notifyNativeEnvironmentSettlement(tx, { companyId: row.run.companyId, runId: row.run.id });
await notifyInstructionCollection(tx, { companyId: row.run.companyId, runId: row.run.id });
await notifyCancelledWorkspacePreservation(tx, { companyId: row.run.companyId, runId: row.run.id });
await notifyUnsafeWorkspaceExportRunChange(tx, row.run);
}
const [updatedRun] = await tx
.update(heartbeatRuns)
.set({
@@ -1038,7 +1117,7 @@ export async function claimNativeRestartRecoveries(input: {
terminalRunToReport = updatedRun;
}
await notifyNativeStatusReassessment(tx, { companyId: row.run.companyId, issueId: row.coordinator.issueId });
await tx
if (!preserveExecutionBinding) await tx
.update(issues)
.set({ executionRunId: null, updatedAt: now })
.where(
@@ -1048,7 +1127,7 @@ export async function claimNativeRestartRecoveries(input: {
eq(issues.executionRunId, row.run.id),
),
);
await tx
if (!preserveExecutionBinding) await tx
.update(issues)
.set({ checkoutRunId: null, updatedAt: now })
.where(
@@ -1059,8 +1138,8 @@ export async function claimNativeRestartRecoveries(input: {
),
);
if (
row.issueAssigneeAgentId === row.run.agentId &&
!["done", "cancelled"].includes(row.issueStatus)
preserveExecutionBinding || (row.issueAssigneeAgentId === row.run.agentId &&
!["done", "cancelled"].includes(row.issueStatus))
)
await issueRecoveryActionService(
tx as unknown as Db,
@@ -1076,17 +1155,21 @@ export async function claimNativeRestartRecoveries(input: {
runId: row.run.id,
reason,
providerAttempt: row.coordinator.attempt,
...(hold ? { automaticAdmissionExhausted: true, executorStopEvidence: "unverified",
providerPid: row.run.processPid, checkpointPreserved: hasCheckpoint } : {}),
},
nextAction:
"Inspect the preserved checkpoint and reconcile the previous execution before starting a fresh session.",
nextAction,
maxAttempts: 3,
wakePolicy: null,
supersedeOnIdentityChange: true,
});
if (updatedRun) await preserveNativeFailedWorkspace(tx as unknown as Db, updatedRun);
if (updatedRun && !preserveExecutionBinding) await preserveNativeFailedWorkspace(tx as unknown as Db, updatedRun);
return { kind: "blocked", runId: row.run.id, reason } as const;
}
const providerAttempt = failedRemoteRetry
? nextNativeProviderAttempt(row.coordinator.attempt)
: row.coordinator.attempt;
const generation = row.coordinator.controllerGeneration + 1;
const leaseOwner = `${controller.bootId}:${generation}:${randomUUID()}`;
const event = historyEntry({
@@ -1097,7 +1180,7 @@ export async function claimNativeRestartRecoveries(input: {
disposition: claimKind,
reason,
requestId: input.recoveryRequestId ?? null,
providerAttempt: row.coordinator.attempt,
providerAttempt,
processPid: row.run.processPid,
processStartedAt: row.run.processStartedAt,
hasCheckpoint,
@@ -1114,6 +1197,7 @@ export async function claimNativeRestartRecoveries(input: {
.update(nativeRunFinalizations)
.set({
phase: "observed",
attempt: providerAttempt,
leaseOwner,
// Cleanup and earlier candidates can outlive the batch observation.
// The owner row is locked: grant the full lease at this write.
@@ -1185,22 +1269,9 @@ export async function claimNativeRestartRecoveries(input: {
updatedAt: now,
})
.where(eq(heartbeatRuns.id, row.run.id));
if (reclaimsExecution) {
const [agent] = await tx.select({ name: agents.name }).from(agents)
.where(eq(agents.id, row.run.agentId)).limit(1);
await tx.update(issues).set({
executionRunId: row.run.id,
executionAgentNameKey: agent?.name.trim().toLowerCase() || null,
executionLockedAt: now,
updatedAt: now,
}).where(and(eq(issues.id, row.coordinator.issueId), eq(issues.companyId, row.run.companyId),
isNull(issues.executionRunId)));
await appendHeartbeatRunEvent(tx as unknown as Db, {
companyId: row.run.companyId, runId: row.run.id, agentId: row.run.agentId,
eventType: "native.recovery.execution_reclaimed", stream: "system", level: "info",
message: "Reclaimed the original task execution for native session recovery.",
payload: { issueId: row.coordinator.issueId },
});
if (releasedRemoteRetry && resumptionAgent) {
await restoreNativeSessionExecution(tx as unknown as Db, { issue, run: row.run, agentName: resumptionAgent.name, now });
}
claimScope = { companyId: row.run.companyId, issueId: row.coordinator.issueId, controller, remote: Boolean(durableAuthority) };
@@ -1208,7 +1279,7 @@ export async function claimNativeRestartRecoveries(input: {
runId: row.run.id,
leaseOwner,
controllerGeneration: generation,
providerAttempt: row.coordinator.attempt,
providerAttempt,
restartKind: input.restartKind,
recoveryRequestId: input.recoveryRequestId ?? null,
};
server/src/services/native-runtime/native-runner-ownership.ts
+36 -2
View File
@@ -1,5 +1,39 @@
import { heartbeatRuns } from "@paperclipai/db";
import { sql } from "drizzle-orm";
import { heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db";
import { and, eq, inArray, isNull, sql } from "drizzle-orm";
import { remoteExecutionHasStopped } from "../remote-execution-termination.js";
import { readSystemRecovery } from "../system-recovery-state.js";
export const nativeManualAdmissionOwnerColumns = {
runId: nativeRunFinalizations.runId, companyId: nativeRunFinalizations.companyId,
issueId: nativeRunFinalizations.issueId, phase: nativeRunFinalizations.phase,
resultId: nativeRunFinalizations.resultId, failureCode: nativeRunFinalizations.failureCode,
failureDetail: sql<Record<string, unknown>>`jsonb_build_object('systemRecovery', ${nativeRunFinalizations.failureDetail}->'systemRecovery')`,
};
type ManualAdmissionOwner = Pick<typeof nativeRunFinalizations.$inferSelect,
"runId" | "companyId" | "issueId" | "phase" | "resultId" | "failureCode" | "failureDetail">;
export function nativeManualAdmissionHoldMatches(scope: { companyId: string; runId: string; issueId?: string },
owner: Partial<ManualAdmissionOwner> | null | undefined): boolean {
const policy = readSystemRecovery(owner?.failureDetail);
return Boolean(owner?.runId === scope.runId && owner.companyId === scope.companyId &&
(!scope.issueId || owner.issueId === scope.issueId) && owner.phase === "terminal_failure" && owner.resultId === null &&
["native_restart_recovery_blocked", "native_retry_cancelled"].includes(owner.failureCode ?? "") &&
policy?.state === "exhausted" && policy.sourceRunId === scope.runId);
}
export async function nativeManualAdmissionHoldsExecution(db: Db, scope: {
companyId: string; runId: string; issueId?: string;
}): Promise<boolean> {
const [hold] = await db.select(nativeManualAdmissionOwnerColumns).from(heartbeatRuns)
.innerJoin(nativeRunFinalizations, and(eq(nativeRunFinalizations.companyId, heartbeatRuns.companyId),
eq(nativeRunFinalizations.runId, heartbeatRuns.id), eq(nativeRunFinalizations.issueId, heartbeatRuns.nativeIssueId)))
.where(and(eq(heartbeatRuns.companyId, scope.companyId), eq(heartbeatRuns.id, scope.runId),
eq(heartbeatRuns.runtimeMode, "native"), scope.issueId ? eq(heartbeatRuns.nativeIssueId, scope.issueId) : undefined,
eq(nativeRunFinalizations.phase, "terminal_failure"), inArray(nativeRunFinalizations.failureCode, ["native_restart_recovery_blocked", "native_retry_cancelled"]),
isNull(nativeRunFinalizations.resultId),
)).limit(1);
return nativeManualAdmissionHoldMatches(scope, hold) && !await remoteExecutionHasStopped(db, scope.companyId, scope.runId);
}
export const NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE =
"native_execution_ownership_unverified";
server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts
+4 -2
View File
@@ -180,7 +180,9 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", (
binding: { companyId, issueId: waiting.issueId, runId: waiting.runId }, bootId: successor.bootId, authority });
} finally { await authority.stop(); }
await waiting.db.update(nativeRunFinalizations).set({ phase: "retryable_failure", attempt: 1,
leaseOwner: null, leaseExpiresAt: null, nextAttemptAt: new Date(0), failureCode: "provider_transport_interrupted" })
leaseOwner: null, leaseExpiresAt: null, nextAttemptAt: new Date(0), failureCode: "provider_transport_interrupted",
failureDetail: { recoveryMode: "exact_checkpoint_resume", checkpointExists: true,
providerSessionEstablished: true, providerEventsExist: false } })
.where(eq(nativeRunFinalizations.runId, waiting.runId));
const state = async () => ({
run: await waiting.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, waiting.runId)),
@@ -222,7 +224,7 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", (
const after = await state();
expect(after.run[0].runnerProfileJson).toEqual(before.run[0].runnerProfileJson);
expect(after.checkpoint).toEqual(before.checkpoint);
expect(after.coordinator[0]).toMatchObject({ attempt: 1, controllerGeneration: 1 });
expect(after.coordinator[0]).toMatchObject({ attempt: 2, controllerGeneration: 1 });
expect(after.task[0].executionRunId).toBe(waiting.runId);
expect(await waiting.db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, waiting.runId),
eq(heartbeatRunEvents.eventType, "native.recovery.execution_reclaimed")))).toHaveLength(1);
server/src/services/native-runtime/native-scheduled-resume-lock.test.ts
+44 -5
View File
@@ -1,4 +1,7 @@
import { randomUUID } from "node:crypto";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { resolve } from "node:path";
import { and, eq } from "drizzle-orm";
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
import {
@@ -9,6 +12,8 @@ import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
import { recoveryService } from "../recovery/service.js";
import { claimNativeRestartRecoveries } from "./native-restart-recovery.js";
import { DurablePrpControlPlane } from "../../vendor/paperclip-runner/index.js";
import { nativeRunnerCheckpointWriter } from "./native-runner-checkpoint.js";
const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_TEST_DATABASE_URL");
const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport();
@@ -19,6 +24,7 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
// run when its retry delay ends.
(support.supported ? describe : describe.skip)("scheduled remote native resume and the task execution lock", () => {
let db: ReturnType<typeof createDb>;
let runtimeRoot: string;
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
beforeAll(async () => {
if (externalUrl) {
@@ -28,10 +34,16 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
temporary = await startEmbeddedPostgresTestDatabase("native-scheduled-resume-lock-");
db = createDb(temporary.connectionString);
}
runtimeRoot = await mkdtemp(resolve(tmpdir(), "scheduled-resume-authority-"));
vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "83".repeat(32));
}, 30_000);
afterAll(async () => { await db?.$client.end(); await temporary?.cleanup(); });
afterAll(async () => {
await db?.$client.end(); await temporary?.cleanup();
if (runtimeRoot) await rm(runtimeRoot, { recursive: true, force: true });
vi.unstubAllEnvs();
});
async function seed(options: { lockedBy?: "self" | "none" } = {}) {
async function seed(options: { lockedBy?: "self" | "none"; durableAuthority?: boolean } = {}) {
const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID();
const sessionId = randomUUID();
const failedAt = new Date(Date.now() - 5 * 60_000);
@@ -64,12 +76,28 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
} }).where(eq(heartbeatRuns.id, runId));
await db.update(issues).set({ executionRunId: lock, checkoutRunId: lock, executionLockedAt: lock ? failedAt : null })
.where(eq(issues.id, issueId));
const owner = { bootId: randomUUID(), generation: 1, leaseOwner: randomUUID() };
await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "retryable_failure", attempt: 1,
controllerBootId: owner.bootId, controllerGeneration: owner.generation,
leaseOwner: owner.leaseOwner, leaseExpiresAt: new Date(Date.now() + 60_000),
nextAttemptAt: new Date(Date.now() - 60_000), recoveryState: "resuming_session",
failureCode: "native_session_interrupted",
failureDetail: { recoveryMode: "exact_checkpoint_resume", checkpointExists: true,
providerSessionEstablished: true, providerEventsExist: false },
});
if (options.durableAuthority !== false) {
const authority = new DurablePrpControlPlane({ stateDirectory: resolve(runtimeRoot, runId),
identity: { runId, normalizedSessionId: sessionId, runnerInstanceId: randomUUID(),
environmentLeaseId: lease.id, turnId: randomUUID(), itemId: randomUUID() },
expectedRunnerVersion: "scheduled-resume-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` });
authority.setCheckpointPersistence(nativeRunnerCheckpointWriter(db, {
companyId, issueId, runId, authorityRunId: runId, ...owner,
}));
await authority.flushCheckpoint();
await authority.stop();
}
await db.update(nativeRunFinalizations).set({ leaseOwner: null, leaseExpiresAt: null })
.where(eq(nativeRunFinalizations.runId, runId));
return { companyId, agentId, issueId, runId };
}
@@ -112,7 +140,7 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
const dispositions = await claimNativeRestartRecoveries({ db, restartKind: "hard", runIds: [f.runId] });
expect(dispositions).toEqual([expect.objectContaining({ kind: "reattach_remote_runner", runId: f.runId })]);
expect(await readIssue(f.issueId)).toMatchObject({ executionRunId: f.runId });
expect(await readCoordinator(f.runId)).toMatchObject({ phase: "observed", attempt: 1, nextAttemptAt: null });
expect(await readCoordinator(f.runId)).toMatchObject({ phase: "observed", attempt: 2, nextAttemptAt: null });
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0])
.toMatchObject({ status: "running", nativePhase: "observed" });
const events = await db.select({ eventType: heartbeatRunEvents.eventType }).from(heartbeatRunEvents)
@@ -120,6 +148,17 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
expect(events).toHaveLength(1);
});
it("keeps a due remote resume held when only the provider snapshot is present", async () => {
const f = await seed({ lockedBy: "none", durableAuthority: false });
expect(await claimNativeRestartRecoveries({ db, restartKind: "hard", runIds: [f.runId] }))
.toMatchObject([{ kind: "blocked", runId: f.runId, reason: "native_retry_remote_authority_unverified" }]);
expect(await readIssue(f.issueId)).toMatchObject({ executionRunId: null, checkoutRunId: null });
expect(await readCoordinator(f.runId)).toMatchObject({ phase: "retryable_failure", attempt: 1 });
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]).toMatchObject({ status: "failed" });
expect(await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, f.runId),
eq(heartbeatRunEvents.eventType, "native.recovery.execution_reclaimed")))).toHaveLength(0);
});
it("does not reclaim a task that another run owns or has admitted", async () => {
for (const competitor of ["lock", "queued"] as const) {
const f = await seed({ lockedBy: "none" });
@@ -128,10 +167,10 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
issueId: f.issueId, status: competitor === "lock" ? "running" : "queued", contextSnapshot: { issueId: f.issueId } });
if (competitor === "lock") await db.update(issues).set({ executionRunId: otherRunId }).where(eq(issues.id, f.issueId));
expect(await claimNativeRestartRecoveries({ db, restartKind: "hard", runIds: [f.runId] })).toEqual([
{ kind: "blocked", runId: f.runId, reason: "issue_execution_lock_changed" },
{ kind: "blocked", runId: f.runId, reason: "native_retry_source_superseded" },
]);
expect((await readIssue(f.issueId)).executionRunId).toBe(competitor === "lock" ? otherRunId : null);
expect(await readCoordinator(f.runId)).toMatchObject({ phase: "retryable_failure", attempt: 1 });
expect(await readCoordinator(f.runId)).toMatchObject({ phase: "terminal_failure", attempt: 1 });
}
});
});
server/src/services/native-runtime/native-session-executor.test.ts
+19 -8
View File
@@ -13,6 +13,7 @@ import { trace } from "@opentelemetry/api";
import { InMemorySpanExporter, SimpleSpanProcessor, TracerProvider } from "@opentelemetry/sdk-trace";
import { configuredEnvironment, configuredEnvironmentProjection } from "../../vendor/paperclip-runner/index.js";
import { afterEach, beforeEach, describe, expect, it, onTestFinished, vi } from "vitest";
import { createSystemRecovery, failSystemRecovery, reserveSystemRecoveryAttempt, SystemRecoveryFailure } from "../system-recovery-state.js";
import { readFileSync } from "node:fs";
import { PgDialect } from "drizzle-orm/pg-core";
import type { SQL } from "drizzle-orm";
@@ -6702,12 +6703,17 @@ function leaseDb(
}
function cancellationDb(options?: {
runId?: string;
coordinator?: {
runId: string;
assessmentId: string | null;
decisionId?: string | null;
phase?: string;
failureCode?: string | null;
companyId?: string;
issueId?: string;
resultId?: string | null;
failureDetail?: Record<string, unknown> | null;
} | null;
status?: string;
failResultJsonUpdateAt?: number;
@@ -6715,7 +6721,7 @@ function cancellationDb(options?: {
resultJson?: Record<string, unknown>;
}) {
const initialRun = {
id: execution.binding.runId,
id: options?.runId ?? execution.binding.runId,
agentId: execution.binding.agentId,
companyId: execution.binding.companyId,
nativeIssueId: execution.binding.issueId,
@@ -6743,7 +6749,7 @@ function cancellationDb(options?: {
const coordinator =
options && "coordinator" in options
? options.coordinator
: { runId: execution.binding.runId, assessmentId: null };
: { runId: initialRun.id, assessmentId: null };
let forUpdateCount = 0;
let resultJsonUpdateCount = 0;
const updates: Array<{ table: unknown; values: Record<string, unknown> }> =
@@ -6795,7 +6801,7 @@ function cancellationDb(options?: {
result.returning = () =>
shouldFail
? Promise.reject(new Error("post_dispatch_db_failure"))
: Promise.resolve([{ id: execution.binding.runId }]);
: Promise.resolve([{ id: initialRun.id }]);
return result;
},
}),
@@ -8464,16 +8470,21 @@ describe("native session cancellation", () => {
expect(state.cancel).not.toHaveBeenCalled();
});
it("fences a failed retry at dispatch and preserves the same audited intent on replay", async () => {
it.each([false, true])("fences a failed retry at dispatch and preserves the same audited intent on replay (manual hold: %s)", async manualHold => {
const id = "11111111-1111-4111-8111-111111111111";
const persistence = cancellationDb({ status: "failed",
coordinator: { runId: execution.binding.runId, assessmentId: null, phase: "retryable_failure" },
const runId = manualHold ? "22222222-2222-4222-8222-222222222222" : execution.binding.runId;
const policy = failSystemRecovery(reserveSystemRecoveryAttempt(createSystemRecovery(runId))!,
new SystemRecoveryFailure("source_binding_invalid", "user_action_required"));
const persistence = cancellationDb({ status: "failed", runId,
coordinator: { runId, assessmentId: null, phase: manualHold ? "terminal_failure" : "retryable_failure",
...(manualHold ? { companyId: execution.binding.companyId, issueId: execution.binding.issueId, resultId: null,
failureCode: "native_restart_recovery_blocked", failureDetail: { systemRecovery: policy } } : {}) },
resultJson: { startupCancellation: { cancellationRequestId: id, retryCancellation: true } } });
const first = await cancelNativeSession(execution.binding.runId, "operator Stop", { db: persistence.db, scope: "run", cancellationRequestId: id });
const first = await cancelNativeSession(runId, "operator Stop", { db: persistence.db, scope: "run", cancellationRequestId: id });
expect(persistence.updates).toContainEqual(expect.objectContaining({ table: nativeRunFinalizations,
values: expect.objectContaining({ phase: "terminal_failure", failureCode: "native_retry_cancelled", nextAttemptAt: null }) }));
const writes = persistence.getResultJsonUpdateCount();
await expect(cancelNativeSession(execution.binding.runId, "retry Stop", { db: persistence.db,
await expect(cancelNativeSession(runId, "retry Stop", { db: persistence.db,
scope: "run", cancellationRequestId: id })).resolves.toMatchObject({ auditId: first.auditId });
expect(persistence.getResultJsonUpdateCount()).toBe(writes);
});
server/src/services/native-runtime/native-session-executor.ts
+9 -4
View File
@@ -266,6 +266,8 @@ import { readProcessStartedAt } from "../hot-restart.js";
import {
NativeRunnerOwnershipUnverifiedError,
isNativeRunnerOwnershipHeld,
nativeManualAdmissionHoldMatches,
nativeManualAdmissionOwnerColumns,
NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE,
NATIVE_ADOPTED_RUNNER_AUTHENTICATION_TIMEOUT,
} from "./native-runner-ownership.js";
@@ -7758,7 +7760,7 @@ async function performNativeCancellation(
const retryCancellation = lockedRun.status === "failed"
|| record(record(lockedRun.resultJson).startupCancellation).retryCancellation === true;
const coordinatorQuery = tx
.select({ runId: nativeRunFinalizations.runId, phase: nativeRunFinalizations.phase, failureCode: nativeRunFinalizations.failureCode })
.select(nativeManualAdmissionOwnerColumns)
.from(nativeRunFinalizations)
.where(
and(
@@ -7785,8 +7787,11 @@ async function performNativeCancellation(
runId, companyId: cancellationContext.companyId, issueId: cancellationContext.issueId,
resultJson, requestId: requestedId, scope: options.scope ?? "run",
}, coordinator)) throw conflict("Native retry is no longer cancellable");
const fenceManualAdmission = coordinator.failureCode === "native_restart_recovery_blocked" && nativeManualAdmissionHoldMatches({
companyId: cancellationContext.companyId, runId, issueId: cancellationContext.issueId,
}, coordinator);
const fenceRetryCancellation = async () => {
if (coordinator.phase !== "retryable_failure") return;
if (coordinator.phase !== "retryable_failure" && !fenceManualAdmission) return;
await tx
.update(nativeRunFinalizations)
.set({
@@ -7802,7 +7807,7 @@ async function performNativeCancellation(
and(
eq(nativeRunFinalizations.runId, runId),
eq(nativeRunFinalizations.companyId, cancellationContext.companyId),
eq(nativeRunFinalizations.phase, "retryable_failure"),
eq(nativeRunFinalizations.phase, coordinator.phase),
),
);
};
@@ -7829,7 +7834,7 @@ async function performNativeCancellation(
: null;
if (!existingAuditId)
throw new Error("native_cancellation_intent_audit_missing");
if (retryCancellation && coordinator.phase === "retryable_failure") {
if (retryCancellation && (coordinator.phase === "retryable_failure" || fenceManualAdmission)) {
await notifyNativeResultCommit(tx, { companyId: cancellationContext.companyId, runId });
await fenceRetryCancellation();
}
server/src/services/native-runtime/native-session-resumption-admission.ts
+73
View File
@@ -0,0 +1,73 @@
import { and, eq, inArray, isNull, ne, or, sql } from "drizzle-orm";
import { agents, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db";
import { evaluateAgentInvokabilityFromDb } from "../agent-invokability.js";
import { issueService } from "../issues.js";
import { isAutomaticRecoverySuppressedByPauseHold } from "../recovery/pause-hold-guard.js";
import { readRunCancellation } from "../run-cancellation.js";
import { getNativeReviewAssignment } from "./native-review-participant.js";
import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js";
import { notifyStatusCardIssueChange } from "../status-card-notifications.js";
import { normalizeAgentNameKey } from "../agent-name-key.js";
export async function restoreNativeSessionExecution(db: Db, input: {
issue: Pick<typeof issues.$inferSelect, "id" | "executionRunId">;
run: Pick<typeof heartbeatRuns.$inferSelect, "id" | "companyId" | "agentId">;
agentName: string;
now: Date;
}) {
const { issue, run, now } = input;
await notifyStatusCardIssueChange(db, run.companyId);
const restored = await db.update(issues).set({ executionRunId: run.id,
executionAgentNameKey: normalizeAgentNameKey(input.agentName), executionLockedAt: now, updatedAt: now,
}).where(and(eq(issues.id, issue.id), eq(issues.companyId, run.companyId),
or(isNull(issues.executionRunId), eq(issues.executionRunId, run.id)),
)).returning({ id: issues.id });
if (!restored.length) throw new Error("native_session_execution_binding_changed");
if (issue.executionRunId !== run.id) await appendHeartbeatRunEvent(db, {
companyId: run.companyId, runId: run.id, agentId: run.agentId,
eventType: "native.recovery.execution_reclaimed", stream: "system", level: "info",
message: "Reclaimed the original task execution for native session recovery.", payload: { issueId: issue.id },
});
}
function nativeSessionResumptionStopped(run: typeof heartbeatRuns.$inferSelect,
coordinator: typeof nativeRunFinalizations.$inferSelect) {
const cancellation = readRunCancellation(run.resultJson);
const nativeCancellation = run.resultJson?.nativeCancellation as { scope?: unknown; dispatchState?: unknown } | undefined;
return (cancellation?.source === "operator" && cancellation.expected) ||
(["run", "issue"].includes(String(nativeCancellation?.scope)) &&
["pending", "acknowledged"].includes(String(nativeCancellation?.dispatchState))) ||
typeof coordinator.failureDetail?.successorRunId === "string";
}
export async function readNativeSessionResumptionAgent(db: Db, input: {
issue: Pick<typeof issues.$inferSelect, "id" | "companyId" | "status" | "executionRunId" | "checkoutRunId" | "assigneeAgentId">;
run: typeof heartbeatRuns.$inferSelect;
coordinator: typeof nativeRunFinalizations.$inferSelect;
}) {
const { issue, run, coordinator } = input;
if (issue.companyId !== run.companyId || issue.id !== run.nativeIssueId ||
coordinator.companyId !== run.companyId || coordinator.issueId !== issue.id || coordinator.runId !== run.id ||
!["todo", "in_progress", "in_review"].includes(issue.status) ||
(issue.executionRunId && issue.executionRunId !== run.id) ||
(issue.checkoutRunId && issue.checkoutRunId !== run.id)) return null;
const [agent] = await db.select().from(agents).where(and(
eq(agents.companyId, run.companyId), eq(agents.id, run.agentId),
)).limit(1);
if (!agent || !(await evaluateAgentInvokabilityFromDb(db, agent)).invokable) return null;
if (issue.assigneeAgentId !== run.agentId && !await getNativeReviewAssignment(db, {
companyId: run.companyId, issueId: issue.id, agentId: run.agentId, contextSnapshot: run.contextSnapshot,
})) return null;
if (await isAutomaticRecoverySuppressedByPauseHold(db, run.companyId, issue.id)) return null;
if (!(await issueService(db).getDependencyReadiness(issue.id, db)).isDependencyReady) return null;
const [competing] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
eq(heartbeatRuns.companyId, run.companyId), ne(heartbeatRuns.id, run.id),
inArray(heartbeatRuns.status, issue.executionRunId === run.id
? ["running", "scheduled_retry"] : ["queued", "running", "scheduled_retry"]),
or(eq(heartbeatRuns.issueId, issue.id), eq(heartbeatRuns.nativeIssueId, issue.id),
sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issue.id}`),
)).limit(1);
if (competing) return null;
if (nativeSessionResumptionStopped(run, coordinator)) return null;
return agent;
}