mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 11:01:23 +01:00
fix(recovery): continue preserved preparation after process loss (#19464)
## Thinking Path > - Paperclip manages AI agents and their execution state. > - A server restart can stop workspace preparation before provider execution. > - Workspace recovery must preserve files before the task can continue. > - Recovery handled interrupted preparations but skipped failed process-loss records with the same no-provider certificate. > - A withheld first retry could also hide the remaining retry budget. > - This change resumes certified preparations through normal admission after file recovery. ## Linked Issues or Issue Description **What happened?** A process-loss attempt remained failed after workspace recovery completed. Its first retry had been cancelled before dispatch by the workspace hold. The task had no running attempt or automatic continuation. **Expected behavior** Continue once after the server proves that provider work did not start, file preservation completed, and the exact leases stopped. Retain the existing failure budget and all task admission gates. **Steps to reproduce** 1. Stop a Runner preparation before native runtime selection. Record the restart recovery bootstrap certificate. 2. Retain the workspace and withhold the first automatic retry before dispatch. 3. Complete file preservation and restart the recovery worker. 4. Observe that the old code schedules no successor. The regression test reproduces this with PostgreSQL and real filesystem operations. **Paperclip version or commit** Reproduced against master at `2f35eb1080`. Related prior work: #19422 continues saved user input after verified copy-back. This change handles automatic continuation of certified process-loss preparation. ## What Changed - Accept failed process-loss preparation only with the durable no-provider bootstrap certificate and existing execution, lease and file checks. - Discover and settle the recovered source through the existing preparation recovery path. - Continue from a verified withheld retry without rewriting its history or resetting its failure count. Adopt it only as the sole retry child; multiple-child history reuses the newest existing child deterministically. - Exercise automatic dispatch, preserved-file readback, one provider turn, accepted result and task completion in the local native-session fixture. - Keep a durable continuation receipt after file-hold settlement until a successor is recorded. Known gates wait for existing eligibility events. Unresolved delivery keeps its shared recovery deadline. - Document the recovery and retry-budget rules. ## Verification - Red: both new PostgreSQL/filesystem regressions failed with zero scheduled successors where one was required. - Green: 54 focused recovery, automatic-dispatch, interruption, multiple-child history and admission safety cases pass on `13c27c2c2f`. The full PostgreSQL admission adapter suite passes all 83 tests; server source declaration typecheck also passes. Local environment: Darwin arm64, Node 26.4.0, PostgreSQL 18.1, Vitest 5.0.3. - Additional red/green: interruption after committed settlement, pause/resume, failed enqueue after cursor advancement, and absence of an earlier retry all failed before their repairs. Restart before enqueue, restart after the successor claim, and pause/resume now preserve one continuation. A rolled-back eligibility change cannot start execution. Deferred and delivered records have no retry deadline. - Local acceptance uses the real heartbeat dispatcher, isolated files and PostgreSQL, with a simulated native provider. It reads the preserved file, writes continued work, accepts the structured result and completes the task once. - Negative cases retain no-certificate, provider-evidence, live-owner, Stop, missing-stop-receipt, wrong-snapshot, human-block, pause, reassignment, approval and exhaustion fences. - CI fixture regression: 4 error-propagation/concurrent-resolution cases pass after adding the new obligation read to their transaction stubs; existing assertions remain unchanged. - Existing idle-drain and notification contract regressions: 4 tests pass after restricting recovery wakes to actual durable obligations. Late approvals block scheduled, queued and final dispatch. - Additional existing retry-scheduling suite: 62 cases pass locally. The pre-provider quota case exceeds its existing 5-second terminal wait on this loaded Darwin host, including when tested alone and with the preceding retry implementation. It remains a local timing limitation; the assertion and timeout were not changed. All Linux CI server shards pass on the combined head, including this unchanged suite. - Existing run-control, legacy-recovery and shared execution-control lifecycle suites: 128 tests pass with the review repair. Required CI supplies final-revision broad coverage. - Real-Codex acceptance was not run: automatic approval review rejected authenticated provider execution because it could incur cost/external effects and required explicit approval. No provider call ran; the controlled-provider acceptance covers the causal restart and durable scheduling behavior. - Hosted acceptance and recovery of existing tasks require rollout. They have not been claimed from the local provider fixture. - Final-head review: Greptile 5/5 on `13c27c2c2f`, no outstanding findings. Another contributor added the multiple-child history safety repair in `13c27c2c2f` while CI was running. The delta was inspected and the 54 focused recovery cases rerun successfully. Required CI passes on that exact combined head: https://github.com/paperclipai/paperclip/actions/runs/42241794918. All 55 successful checks include aggregate `ci / verify`, typecheck, build, canary, all server and E2E shards, and Runner package/Rust/static checks. The separate Paperclip Review service check remains queued waiting for its assigned agent; no validated assessment has been returned. ## Risks A malformed preparation certificate or a false no-execution receipt could permit duplicate work. The existing checks still reject provider execution records, native coordinators, live controllers, unsettled leases and unverified files. The pending continuation must survive a restart without polling known admission blockers. A withheld retry retains its original row and consumes its remaining failure budget. Normal admission checks current ownership, approvals, pause and budgets. ## Model Used OpenAI GPT-6 family via Codex. The exact model ID and context window are unavailable in session metadata. Used reasoning, code editing and tool execution. ## 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 task 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> Co-authored-by: Devin Foley <devin@paperclip.ing>
This commit is contained in:
12 files changed
+413
-40
No files matched your search
@@ -2589,13 +2589,26 @@ Shutdown, process loss, and provider failure use the existing durable failure re
|
||||
If shutdown or a recorded unexpected system interruption stops Paperclip Runner
|
||||
preparation before native execution was selected, recovery verifies that no
|
||||
provider, process launch, native result or coordinator exists. The previous
|
||||
controller must have settled or lost its lease.
|
||||
controller must have settled or lost its lease. A failed `process_lost` run also
|
||||
qualifies when restart recovery committed its explicit bootstrap certificate
|
||||
that provider work did not start. The error label alone grants no replay.
|
||||
Required workspace recovery must have an exact saved-file receipt and stopped
|
||||
leases before the task can continue. Once these checks pass, the server retires
|
||||
its own unchanged bootstrap-failure block and schedules one bounded retry.
|
||||
Startup also discovers this obligation when file recovery completed before the
|
||||
restart. A pending Retry or saved message owns its continuation instead of
|
||||
starting a competing automatic attempt. Retry uses the same preparation proof
|
||||
starting a competing automatic attempt. If a replacement was already withheld
|
||||
by the workspace gate before execution, continuation retains that cancelled row
|
||||
and uses its existing predecessor claim and remaining failure budget.
|
||||
Recovery adopts that withheld row only when it is the source's sole retry child.
|
||||
Multiple-child history reuses the newest existing child, ordered by creation time
|
||||
and run ID, without scheduling another replacement.
|
||||
Clearing the file hold records a pending continuation before enqueue. Restart
|
||||
reconstructs that obligation until a successor is recorded. Known admission
|
||||
gates defer it without a polling deadline; eligibility changes wake the existing
|
||||
recovery worker. An unresolved enqueue or receipt write keeps its shared retry
|
||||
deadline. Delivery or exhaustion retires that obligation.
|
||||
Retry uses the same preparation proof
|
||||
and preserves the recovered workspace. Current task ownership, human blocks,
|
||||
pauses, approvals, budgets and the existing retry limit remain effective.
|
||||
Operator Stops retain their explicit continuation requirements. Other ambiguous
|
||||
|
||||
@@ -99,6 +99,7 @@ describe("issueRecoveryActionService", () => {
|
||||
limit() { return Promise.resolve([]); } };
|
||||
const fakeDb = {
|
||||
async transaction<T>(fn: (tx: unknown) => Promise<T>): Promise<T> { return fn(fakeDb); },
|
||||
execute: vi.fn(async () => [{ pendingPreparationContinuation: false }]),
|
||||
select: vi.fn(() => query),
|
||||
insert: vi.fn(() => ({ values: () => ({ returning: async () => { throw error; } }) })),
|
||||
};
|
||||
@@ -130,6 +131,7 @@ describe("issueRecoveryActionService", () => {
|
||||
|
||||
const fakeDb = {
|
||||
async transaction<T>(fn: (tx: unknown) => Promise<T>): Promise<T> { return fn(fakeDb); },
|
||||
execute: vi.fn(async () => [{ pendingPreparationContinuation: false }]),
|
||||
select: vi.fn(() => makeSelectQuery(selectResults.shift() ?? [])),
|
||||
update: vi.fn(() => ({
|
||||
set: vi.fn(() => ({
|
||||
|
||||
@@ -20,6 +20,7 @@ import {
|
||||
environmentLeases,
|
||||
heartbeatRunEvents,
|
||||
heartbeatRuns,
|
||||
issueApprovals,
|
||||
issueDocuments,
|
||||
issueRelations,
|
||||
issueRecoveryActions,
|
||||
@@ -379,6 +380,33 @@ describeEmbeddedPostgres("run-dispatch postgres adapter", () => {
|
||||
expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]!.status).toBe("blocked");
|
||||
});
|
||||
|
||||
it("keeps recovered preparation behind approvals added after scheduling", async () => {
|
||||
const { companyId, agentId } = await seedCompanyAndAgent();
|
||||
const issueId = randomUUID();
|
||||
await seedIssue({ companyId, issueId, assigneeAgentId: agentId, status: "in_progress" });
|
||||
const retryReason = "interrupted_preparation";
|
||||
const replacementId = await seedRun({ companyId, agentId, status: "scheduled_retry",
|
||||
contextSnapshot: { issueId, wakeReason: retryReason, retryReason, forceFreshSession: true } });
|
||||
const adapter = createPostgresRunDispatchAdapter(db);
|
||||
expect(await adapter.evaluateScheduledRetryGate({ companyId, runId: replacementId, retryReasonOverride: retryReason, now: new Date() }))
|
||||
.toMatchObject({ allowed: true });
|
||||
const approvalId = randomUUID();
|
||||
await db.insert(approvals).values({ id: approvalId, companyId, type: "hire_agent", status: "pending", payload: {} });
|
||||
await db.insert(issueApprovals).values({ companyId, issueId, approvalId });
|
||||
expect(await adapter.evaluateScheduledRetryGate({ companyId, runId: replacementId, retryReasonOverride: retryReason, now: new Date() }))
|
||||
.toMatchObject({ allowed: false, errorCode: "issue_waiting_for_response" });
|
||||
await db.update(heartbeatRuns).set({ status: "queued" }).where(eq(heartbeatRuns.id, replacementId));
|
||||
const reservation = await reserveRunBudget(db, companyId, replacementId, null);
|
||||
expect(await adapter.cancelStaleQueuedRun({ companyId, runId: replacementId, expectedStatus: "queued", now: new Date() }))
|
||||
.toMatchObject({ outcome: "cancelled", errorCode: "issue_waiting_for_response" });
|
||||
await db.update(heartbeatRuns).set({ status: "running" }).where(eq(heartbeatRuns.id, replacementId));
|
||||
const dispatch = vi.fn(async () => undefined);
|
||||
expect(await adapter.dispatchResolvedInteractionIfCurrent({ companyId, runId: replacementId,
|
||||
reservationId: reservation.id, expectedStatus: "running", now: new Date(), dispatch }))
|
||||
.toMatchObject({ dispatched: false, cancellation: { outcome: "cancelled" } });
|
||||
expect(dispatch).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it.each(["queued", "final", "resolved"] as const)("rechecks late native replacement dependencies at %s dispatch", async mode => {
|
||||
const { companyId, agentId } = await seedCompanyAndAgent();
|
||||
const issueId = randomUUID(), blockerId = randomUUID();
|
||||
|
||||
@@ -417,7 +417,7 @@ export function createPostgresRunDispatchAdapter(
|
||||
facts.issueAssigneeAgentId = issue.assigneeAgentId;
|
||||
facts.issueExecutionRunId = issue.executionRunId;
|
||||
facts.issueCheckoutRunId = issue.checkoutRunId;
|
||||
if (input.conversationContinuation) {
|
||||
if (input.conversationContinuation || retryReason === "interrupted_preparation") {
|
||||
facts.pendingResponse = await connectionContinuationPendingResponse(dbOrTx as unknown as Db,
|
||||
issue, input.responsibleUserId, input.runId);
|
||||
}
|
||||
@@ -657,7 +657,7 @@ export function createPostgresRunDispatchAdapter(
|
||||
unresolvedBlockerIssueIds: readiness.unresolvedBlockerIssueIds,
|
||||
unresolvedBlockerCount: readiness.unresolvedBlockerCount,
|
||||
} : null,
|
||||
pendingResponse: issue && (input.conversationContinuation || context.connectionIntentResolution === "existing_connection")
|
||||
pendingResponse: issue && (input.conversationContinuation || retryReason === "interrupted_preparation" || context.connectionIntentResolution === "existing_connection")
|
||||
? await connectionContinuationPendingResponse(dbOrTx as unknown as Db, issue, input.responsibleUserId, input.runId)
|
||||
: null,
|
||||
issueFound: issue !== null,
|
||||
|
||||
@@ -103,14 +103,18 @@ function hasOnlyPreparationAccountingContext(usage: unknown, issueId: unknown):
|
||||
* stopped leases, and any saved-file obligation below. */
|
||||
export function isInterruptedNativePreparationCandidate(run: Run): boolean {
|
||||
const cancellation = readRunCancellation(run.resultJson);
|
||||
const recovery = run.resultJson?.executionRecovery as Record<string, unknown> | undefined;
|
||||
// Restart recovery records this certificate only after revoking a preparing
|
||||
// controller. A generic process-lost error cannot establish safe replay.
|
||||
const lostPreparation = run.status === "failed" && run.errorCode === "process_lost"
|
||||
&& recovery?.kind === "bootstrap" && recovery.providerWorkStarted === false;
|
||||
// Older shutdown records predate structured cancellation. Current records
|
||||
// identify unexpected system interruption without relying on an error label.
|
||||
const systemInterruption = cancellation
|
||||
? cancellation.initiator.type === "system" && cancellation.source !== "operator" &&
|
||||
(cancellation.source === "shutdown" || cancellation.expected === false)
|
||||
: run.errorCode === "server_shutdown_interrupted";
|
||||
const recovery = run.resultJson?.executionRecovery as Record<string, unknown> | undefined;
|
||||
return Boolean(systemInterruption && run.status === "interrupted" && run.finishedAt
|
||||
: run.errorCode === "server_shutdown_interrupted" || lostPreparation;
|
||||
return Boolean(systemInterruption && (run.status === "interrupted" || lostPreparation) && run.finishedAt
|
||||
&& run.runtimeMode === "legacy" && !run.runtimeModeResolvedAt && run.executionStage === "preparing"
|
||||
&& claimedAdapterType(run) === "paperclip_runner" && !run.processPid && !run.processGroupId
|
||||
&& !run.processStartedAt && !run.sessionIdAfter && !run.nativeIssueId && !run.nativeSessionId
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { sql } from "drizzle-orm";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
|
||||
export type DeferredExecutionScope = { companyId: string; requestId?: string };
|
||||
@@ -24,6 +26,17 @@ export async function notifyTaskRetrySettlement(tx: object, scope: DeferredExecu
|
||||
}
|
||||
|
||||
export async function notifyDeferredExecutionEligibility(tx: object, scope?: { companyId: string }) {
|
||||
// Only an actual continuation obligation may wake this passive worker.
|
||||
// Empty eligibility hints during idle drain must not keep maintenance active.
|
||||
const pending = await (tx as Pick<Db, "execute">).execute<{ pendingPreparationContinuation: boolean }>(sql`
|
||||
select exists (select 1 from issue_recovery_actions
|
||||
where evidence ? 'systemRecovery' and evidence#>>'{systemRecovery,state}' = 'complete'
|
||||
and evidence->>'bootstrapPreparationSettledAt' is not null
|
||||
and coalesce(evidence->>'bootstrapPreparationContinuation', 'pending') in ('pending', 'deferred')
|
||||
${scope ? sql`and company_id = ${scope.companyId}::uuid` : sql``}
|
||||
) as "pendingPreparationContinuation"`);
|
||||
if (pending[0]?.pendingPreparationContinuation === true)
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.workspaceRecovery, scope ? JSON.stringify({ companyId: scope.companyId }) : "*");
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeProcessSessions, scope ? JSON.stringify({ companyId: scope.companyId }) : "*");
|
||||
if (scope) {
|
||||
await notifyDeferredExecutionWork(tx, scope);
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import { isInterruptedNativePreparation } from "../cancelled-native-startup.js";
|
||||
import { isPreDispatchExecutionWaitVerified } from "../pre-dispatch-execution-wait.js";
|
||||
import { notifyAttentionRetentionWork } from "../retention-work-notifications.js";
|
||||
import { normalizeAgentNameKey } from "../agent-name-key.js";
|
||||
import { notifyStatusCardIssueChange } from "../status-card-notifications.js";
|
||||
@@ -430,6 +432,20 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen
|
||||
delayMs?: number;
|
||||
},
|
||||
) {
|
||||
const preparation = parseObject(run.resultJson?.executionRecovery);
|
||||
if (typeof preparation.preparationSettledAt === "string" && await isInterruptedNativePreparation(db, run)) {
|
||||
const children = await db.select().from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.agentId, run.agentId),
|
||||
eq(heartbeatRuns.retryOfRunId, run.id),
|
||||
)).orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)).limit(2);
|
||||
const withheld = children.length === 1 ? children[0] : undefined;
|
||||
if (withheld && await isPreDispatchExecutionWaitVerified(db, withheld)) {
|
||||
// The first replacement was withheld before execution while files were
|
||||
// recovered. Preserve its history, budget and predecessor claim.
|
||||
run = withheld;
|
||||
opts = { ...opts, retryReason: opts?.retryReason ?? "interrupted_preparation" };
|
||||
}
|
||||
}
|
||||
if (!opts?.retryReason && run.errorCode === RUNNER_REVISION_MISMATCH_CODE) opts = { ...RUNNER_REVISION_MISMATCH_RETRY, ...opts };
|
||||
if (await hasCaseHandoffForRun(run)) return { outcome: "not_scheduled" as const,
|
||||
reason: "The durable case handoff owns successor work.", issueId: readNonEmptyString(run.contextSnapshot?.issueId) };
|
||||
@@ -449,8 +465,9 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen
|
||||
}
|
||||
const now = opts?.now ?? new Date();
|
||||
const preparationRecovery = parseObject(run.resultJson?.executionRecovery);
|
||||
const interruptedPreparation = run.status === "interrupted" &&
|
||||
preparationRecovery.kind === "bootstrap" && preparationRecovery.providerWorkStarted === false;
|
||||
const interruptedPreparation = ["interrupted", "failed"].includes(run.status) &&
|
||||
preparationRecovery.kind === "bootstrap" && preparationRecovery.providerWorkStarted === false
|
||||
&& (run.status === "interrupted" || await isInterruptedNativePreparation(db, run));
|
||||
const retryReason = opts?.retryReason ??
|
||||
(interruptedPreparation ? "interrupted_preparation" : BOUNDED_TRANSIENT_HEARTBEAT_RETRY_REASON);
|
||||
const wakeReason =
|
||||
@@ -590,7 +607,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen
|
||||
: baseSchedule;
|
||||
|
||||
const requiresIssueGate =
|
||||
interruptedPreparation ||
|
||||
interruptedPreparation || retryReason === "interrupted_preparation" ||
|
||||
(run.status === "interrupted" && run.errorCode === "server_shutdown_interrupted") ||
|
||||
isTransientWorkspacePreparationCode(run.errorCode) ||
|
||||
hasConversationContinuationPolicy(run.resultJson) ||
|
||||
@@ -766,6 +783,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen
|
||||
eq(heartbeatRuns.retryOfRunId, run.id),
|
||||
),
|
||||
)
|
||||
.orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id))
|
||||
.limit(1);
|
||||
if (existing)
|
||||
return {
|
||||
|
||||
@@ -55,7 +55,7 @@ import {
|
||||
undeliveredLegacyUserCommentIds,
|
||||
} from "../explicit-native-continuation.js";
|
||||
import { isCancelledNativeStartup, isInterruptedNativePreparation } from "../cancelled-native-startup.js";
|
||||
import { settleInterruptedNativeBootstrap } from "../legacy-execution-recovery.js";
|
||||
import { settleInterruptedNativeBootstrap, recordInterruptedPreparationContinuation } from "../legacy-execution-recovery.js";
|
||||
import {
|
||||
executionBlockerPredicate,
|
||||
getExecutionBlocker,
|
||||
@@ -573,7 +573,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont
|
||||
async function resumeRemoteStopComments(run: NonNullable<Awaited<ReturnType<typeof getRunExecutionWait>>>, requestId?: string) {
|
||||
if (!isHeartbeatRunTerminalStatus(run.status) || adapterExecutionControls.has(run.id)) return;
|
||||
let interruptedPreparation = false;
|
||||
if (run.status === "interrupted" && run.runtimeMode === "legacy" &&
|
||||
if (["interrupted", "failed"].includes(run.status) && run.runtimeMode === "legacy" &&
|
||||
!run.runtimeModeResolvedAt && run.executionStage === "preparing") {
|
||||
const current = await getRun(run.id, { includeExecutionEvidence: true });
|
||||
if (current && await isInterruptedNativePreparation(db, current)) {
|
||||
@@ -617,12 +617,20 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont
|
||||
requestId ? eq(agentWakeupRequests.id, requestId) : undefined,
|
||||
sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`,
|
||||
)).orderBy(asc(agentWakeupRequests.requestedAt)).limit(50);
|
||||
if (interruptedPreparation && !pending.length && !requestId && !(await getSchedulingSuppression()).suppressed) {
|
||||
const [task] = await db.select({ status: issues.status, assigneeAgentId: issues.assigneeAgentId }).from(issues).where(and(
|
||||
eq(issues.companyId, run.companyId), eq(issues.id, issueId),
|
||||
));
|
||||
if (task?.assigneeAgentId === run.agentId && ["todo", "in_progress", "in_review"].includes(task.status)) {
|
||||
await dependencies.scheduleBoundedRetry(run.id);
|
||||
if (interruptedPreparation) {
|
||||
if (pending.length || requestId) await recordInterruptedPreparationContinuation(db, run, "delegated");
|
||||
else {
|
||||
const [task] = await db.select({ status: issues.status, assigneeAgentId: issues.assigneeAgentId }).from(issues).where(and(
|
||||
eq(issues.companyId, run.companyId), eq(issues.id, issueId),
|
||||
));
|
||||
if (!(await getSchedulingSuppression()).suppressed && task?.assigneeAgentId === run.agentId &&
|
||||
["todo", "in_progress", "in_review"].includes(task.status)) {
|
||||
const retry = await dependencies.scheduleBoundedRetry(run.id);
|
||||
await recordInterruptedPreparationContinuation(db, run,
|
||||
retry.outcome === "scheduled" ? retry.run.status === "cancelled" ? "blocked" : "delivered"
|
||||
: retry.outcome === "retry_exhausted" ? "exhausted" : "deferred",
|
||||
retry.outcome === "scheduled" ? retry.run.id : undefined);
|
||||
} else await recordInterruptedPreparationContinuation(db, run, "deferred");
|
||||
}
|
||||
}
|
||||
for (const wake of pending) {
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
|
||||
import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
|
||||
import { notifyInstructionCollection } from "./instruction-collection-notifications.js";
|
||||
@@ -96,7 +97,7 @@ export function legacyExecutionNeedsReconciliation(
|
||||
/** Admission-wait receipts are only exempt after retained execution evidence agrees.
|
||||
* The synchronous classifier stays conservative for callers without a DB proof. */
|
||||
export async function legacyExecutionNeedsReconciliationWithEvidence(db: Db, run: Run): Promise<boolean> {
|
||||
if (run.status === "interrupted" && await isInterruptedNativePreparation(db, run)) return false;
|
||||
if (await isInterruptedNativePreparation(db, run)) return false;
|
||||
return await hasUnrestoredRemoteWorkspace(db, run) ||
|
||||
(legacyExecutionNeedsReconciliation(run) && !(await isPreDispatchReviewWaitVerified(db, run))
|
||||
&& !(await isPreDispatchExecutionWaitVerified(db, run)));
|
||||
@@ -369,14 +370,17 @@ export async function settleInterruptedNativeBootstrap(
|
||||
const [settled] = await tx.update(heartbeatRuns).set({ resultJson: { ...run.resultJson,
|
||||
executionRecovery: { kind: "bootstrap", providerWorkStarted: false, preparationSettledAt: now.toISOString() },
|
||||
}, updatedAt: now }).where(eq(heartbeatRuns.id, run.id)).returning();
|
||||
if (holds.length) await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: task.companyId, issueId: task.id });
|
||||
if (holds.length) {
|
||||
await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: task.companyId, issueId: task.id });
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.workspaceRecovery);
|
||||
}
|
||||
const resolved = holds.length ? await tx.update(issueRecoveryActions).set({ status: "resolved", outcome: "false_positive",
|
||||
resolutionNote: "The interrupted preparation has no provider execution and its cleanup is complete.",
|
||||
resolvedAt: now, updatedAt: now,
|
||||
// An automatic disposition may already have resolved the incident but
|
||||
// left replay blocked. Preserve its history while correcting that verdict.
|
||||
evidence: sql`jsonb_set(${issueRecoveryActions.evidence}, '{automaticRecovery,replay}', '"allowed"'::jsonb, false) ||
|
||||
${JSON.stringify({ bootstrapPreparationSettledAt: now.toISOString() })}::jsonb`,
|
||||
${JSON.stringify({ bootstrapPreparationSettledAt: now.toISOString(), bootstrapPreparationContinuation: "pending" })}::jsonb`,
|
||||
}).where(inArray(issueRecoveryActions.id, holds.map(hold => hold.id))).returning({ id: issueRecoveryActions.id }) : [];
|
||||
if (resolved.length) await logActivity(tx as unknown as Db, { companyId: run.companyId, actorType: "system", actorId: "system",
|
||||
action: "issue.recovery_action_resolved", entityType: "issue", entityId: task.id, agentId: run.agentId, runId: run.id,
|
||||
@@ -386,6 +390,29 @@ export async function settleInterruptedNativeBootstrap(
|
||||
});
|
||||
}
|
||||
|
||||
/** The source stays discoverable until normal admission records a successor.
|
||||
* Known admission gates wait for eligibility events; unresolved writes retain
|
||||
* the pending delivery deadline. Concurrent successful delivery never regresses. */
|
||||
export async function recordInterruptedPreparationContinuation(db: Db, run: Pick<Run, "id" | "companyId">,
|
||||
state: "deferred" | "delivered" | "delegated" | "exhausted" | "blocked", successorRunId?: string) {
|
||||
return db.transaction(async tx => {
|
||||
const actions = await tx.select().from(issueRecoveryActions).where(and(
|
||||
eq(issueRecoveryActions.companyId, run.companyId), eq(issueRecoveryActions.fingerprint, `legacy-execution:${run.id}`),
|
||||
sql`${issueRecoveryActions.evidence}->>'runId' = ${run.id}`,
|
||||
sql`${issueRecoveryActions.evidence}->>'bootstrapPreparationSettledAt' is not null`,
|
||||
)).for("update");
|
||||
const pending = actions.filter(action =>
|
||||
!["delivered", "delegated", "exhausted", "blocked"].includes(String(action.evidence.bootstrapPreparationContinuation))
|
||||
&& action.evidence.bootstrapPreparationContinuation !== state);
|
||||
if (!pending.length) return;
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.workspaceRecovery);
|
||||
for (const action of pending) await tx.update(issueRecoveryActions).set({ updatedAt: new Date(), evidence: {
|
||||
...action.evidence, bootstrapPreparationContinuation: state,
|
||||
...(successorRunId ? { bootstrapPreparationContinuationRunId: successorRunId } : {}),
|
||||
} }).where(eq(issueRecoveryActions.id, action.id));
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* The executor that owned a cancelled legacy run proves that its gated
|
||||
* dispatch never entered the adapter. Such a Stop (for example a subtree pause
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { heartbeatRuns, issueComments, issueRecoveryActions, type Db } from "@paperclipai/db";
|
||||
import { agents, issues, heartbeatRuns, issueComments, issueRecoveryActions, type Db } from "@paperclipai/db";
|
||||
import { writeExecutionControl } from "./execution-control-notifications.js";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
import { createSystemRecovery, readSystemRecovery, type SystemRecoveryState } from "./system-recovery-state.js";
|
||||
@@ -127,8 +127,18 @@ export async function nextWorkspaceRecoveryAt(db: Db): Promise<number | null> {
|
||||
then (${issueRecoveryActions.evidence}->>'runId')::uuid end`),
|
||||
sql`${heartbeatRuns.status} in ('failed', 'interrupted', 'timed_out', 'cancelled', 'succeeded')`,
|
||||
)).where(and(
|
||||
workspaceRecoverySourceEligible(),
|
||||
sql`${issueRecoveryActions.evidence}#>>'{systemRecovery,state}' in ('scheduled', 'running', 'awaiting_evidence')`,
|
||||
workspaceRecoverySourceEligible(true),
|
||||
sql`(${issueRecoveryActions.evidence}#>>'{systemRecovery,state}' in ('scheduled', 'running', 'awaiting_evidence')
|
||||
or (${issueRecoveryActions.evidence}#>>'{systemRecovery,state}' = 'complete'
|
||||
and ${issueRecoveryActions.evidence}->>'bootstrapPreparationSettledAt' is not null
|
||||
and coalesce(${issueRecoveryActions.evidence}->>'bootstrapPreparationContinuation', 'pending') = 'pending'
|
||||
and exists (select 1 from ${issues} inner join ${agents}
|
||||
on ${agents.id} = ${heartbeatRuns.agentId} and ${agents.companyId} = ${heartbeatRuns.companyId}
|
||||
where ${issues.id} = ${issueRecoveryActions.sourceIssueId} and ${issues.companyId} = ${heartbeatRuns.companyId}
|
||||
and ${issues.assigneeAgentId} = ${heartbeatRuns.agentId}
|
||||
and ${issues.status} in ('todo', 'in_progress', 'in_review')
|
||||
and ${issues.executionRunId} is null and ${issues.checkoutRunId} is null
|
||||
and ${agents.status} in ('idle', 'running', 'error'))))`,
|
||||
)).orderBy(sql`1`).limit(1);
|
||||
return rows.length ? Number(rows[0].at) : null;
|
||||
}
|
||||
@@ -1,3 +1,5 @@
|
||||
import { nextWorkspaceRecoveryAt } from "./system-recovery.js";
|
||||
import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js";
|
||||
import { createAdmissionTransactionScope, createWakeAdmissionReader, createWakeAdmissionWriter } from "../modules/wake-queue/adapters/postgres.js";
|
||||
import { createAdmitWakeBehindIssueExecution } from "../modules/wake-queue/application/use-cases.js";
|
||||
import { filterZombieCoalesceTarget } from "./heartbeat.js";
|
||||
@@ -9,7 +11,7 @@ import { isPreDispatchExecutionWaitVerified } from "./pre-dispatch-execution-wai
|
||||
import { heartbeatRunEvents } from "@paperclipai/db";
|
||||
import { createHeartbeatQueue, shouldDeferFollowupWakeForSameIssue, shouldQueueFollowupForRunningIssueWake } from "./heartbeat/queue.js";
|
||||
import { createHeartbeatQueueTestDependencies } from "../__tests__/helpers/heartbeat-queue.js";
|
||||
import { recordRequiredWorkspaceRestoreFailure, terminalizeLegacyExecution } from "./legacy-execution-recovery.js";
|
||||
import { recordRequiredWorkspaceRestoreFailure, terminalizeLegacyExecution, settleInterruptedNativeBootstrap } from "./legacy-execution-recovery.js";
|
||||
import { admitExplicitNativeContinuation } from "./explicit-native-continuation.js";
|
||||
import { withQueuedCommentIdsInWakePayload } from "./issue-queued-comment-queue.js";
|
||||
import * as recoverySources from "./workspace-recovery-source.js";
|
||||
@@ -1451,17 +1453,20 @@ describe("automatic retained workspace recovery", () => {
|
||||
expect(await readFile(path.join(f.remote, "saved.txt"), "utf8")).toBe("finished work\n");
|
||||
}, 30_000);
|
||||
|
||||
async function interruptedRemotePreparation(interruption: "shutdown" | "lease" = "shutdown", sourceLayout: RecoverySourceLayout = "retained") {
|
||||
async function interruptedRemotePreparation(interruption: "shutdown" | "lease" | "process_lost" = "shutdown", sourceLayout: RecoverySourceLayout = "retained") {
|
||||
const f = await acceptedRemoteExport(sourceLayout);
|
||||
await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
await db.delete(nativeRunResults).where(eq(nativeRunResults.runId, f.runId));
|
||||
const recovery = { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [f.lease.id] };
|
||||
await db.update(heartbeatRuns).set({ runtimeMode: "legacy", nativeIssueId: null, completionContractId: null,
|
||||
status: "interrupted", executionStage: "preparing",
|
||||
errorCode: interruption === "lease" ? "lease_released_before_terminal" : "server_shutdown_interrupted",
|
||||
status: interruption === "process_lost" ? "failed" : "interrupted", executionStage: "preparing",
|
||||
errorCode: interruption === "process_lost" ? "process_lost" : interruption === "lease" ? "lease_released_before_terminal" : "server_shutdown_interrupted",
|
||||
controllerBootId: randomUUID(),
|
||||
error: interruption === "lease" ? "Environment lease released before preparation settled" : "Interrupted by graceful server shutdown (SIGTERM)",
|
||||
error: interruption === "process_lost" ? "Process lost -- server may have restarted; retrying once"
|
||||
: interruption === "lease" ? "Environment lease released before preparation settled" : "Interrupted by graceful server shutdown (SIGTERM)",
|
||||
resultJson: { workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: recovery,
|
||||
...(interruption === "process_lost" ? { executionRecovery: { kind: "bootstrap", providerWorkStarted: false },
|
||||
processLossDiagnostic: { localCheck: "no_identifiers", pidRecorded: false, groupRecorded: false, retryEligible: true, runPredatesObserver: true } } : {}),
|
||||
...(interruption === "lease" ? { cancellation: { source: "unknown", expected: false,
|
||||
initiator: { type: "system" }, reason: "Environment lease released before preparation settled",
|
||||
recordedAt: new Date().toISOString() } } : {}) },
|
||||
@@ -1478,6 +1483,18 @@ describe("automatic retained workspace recovery", () => {
|
||||
return f;
|
||||
}
|
||||
|
||||
async function withheldPreparationRetry(f: { companyId: string; issueId: string; runId: string; actionId: string }) {
|
||||
const source = await run(f.runId);
|
||||
const [withheld] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: source.agentId,
|
||||
responsibleUserId: source.responsibleUserId, scopeKind: "issue", issueId: f.issueId, runtimeMode: "legacy", status: "cancelled", finishedAt: new Date(),
|
||||
retryOfRunId: f.runId, scheduledRetryAttempt: 1, scheduledRetryReason: "transient_failure",
|
||||
errorCode: "execution_reconciliation_required", contextSnapshot: { issueId: f.issueId },
|
||||
resultJson: { stopReason: "execution_reconciliation_required", timeoutSource: "stale_queued_run_gate",
|
||||
executionWait: { issueId: f.issueId, recoveryActionId: f.actionId } },
|
||||
}).returning();
|
||||
return withheld;
|
||||
}
|
||||
|
||||
async function failedRemotePreparation(sourceLayout: RecoverySourceLayout = "retained", startedMarker = false, warning = false) {
|
||||
const f = await interruptedRemotePreparation("shutdown", sourceLayout);
|
||||
const source = await run(f.runId);
|
||||
@@ -2016,9 +2033,9 @@ describe("automatic retained workspace recovery", () => {
|
||||
expect(await readFile(path.join(cwd, "saved.txt"), "utf8")).toBe("existing task work\n");
|
||||
});
|
||||
|
||||
it.each(["complete", "catalog", "catalog_warning", "catalog_unknown_warning", "catalog_provider_warning", "invalid_published_root"] as const)(
|
||||
it.each(["complete", "automatic_lost", "catalog", "catalog_warning", "catalog_unknown_warning", "catalog_provider_warning", "invalid_published_root"] as const)(
|
||||
"dispatches the recovered failed-preparation Retry once through the native provider boundary (%s)", async mode => {
|
||||
const unavailableCatalog = mode !== "complete" && mode !== "invalid_published_root", withWarning = mode.includes("warning");
|
||||
const unavailableCatalog = mode !== "complete" && mode !== "automatic_lost" && mode !== "invalid_published_root", withWarning = mode.includes("warning");
|
||||
const localStorage = getStorageService();
|
||||
const databaseUrl = process.env.DATABASE_URL, masterKey = process.env.PAPERCLIP_SECRETS_MASTER_KEY;
|
||||
const experimental = await instanceSettingsService(db).getExperimental();
|
||||
@@ -2036,7 +2053,12 @@ describe("automatic retained workspace recovery", () => {
|
||||
storage = vi.spyOn(storageBackend, "getStorageService").mockReturnValue({ ...localStorage, provider: "s3" });
|
||||
process.env.DATABASE_URL = connectionString; process.env.PAPERCLIP_SECRETS_MASTER_KEY = "63".repeat(32);
|
||||
await instanceSettingsService(db).updateExperimental({ enableNativeRunner: true, enableIsolatedWorkspaces: true });
|
||||
const f = await failedRemotePreparation(mode === "invalid_published_root" ? "raw" : "published");
|
||||
const f = mode === "automatic_lost" ? await interruptedRemotePreparation("process_lost", "published")
|
||||
: await failedRemotePreparation(mode === "invalid_published_root" ? "raw" : "published");
|
||||
if (mode === "automatic_lost") {
|
||||
await db.update(heartbeatRuns).set({ responsibleUserId: "local-board" }).where(eq(heartbeatRuns.id, f.runId));
|
||||
await withheldPreparationRetry(f);
|
||||
}
|
||||
if (mode === "complete") await db.update(heartbeatRuns).set({ sessionIdBefore: randomUUID() }).where(eq(heartbeatRuns.id, f.runId));
|
||||
await recoverRetainedWorkspace(f.input);
|
||||
const source = await run(f.runId), nextDirectory = path.join(root, randomUUID(), "next");
|
||||
@@ -2144,12 +2166,19 @@ describe("automatic retained workspace recovery", () => {
|
||||
const app = boardRecoveryApp(undefined, heartbeat);
|
||||
const gate = new Promise<void>(resolve => { releaseStart = resolve; });
|
||||
heldStart = agentStarts.withAgentStartLock(source.agentId, () => gate);
|
||||
const receipt = (await request(app).post(`/api/issues/${f.issueId}/retry`)
|
||||
const receipt = mode === "automatic_lost" ? null : (await request(app).post(`/api/issues/${f.issueId}/retry`)
|
||||
.send({ observedRunId: f.runId, idempotencyKey: randomUUID() }).expect(202)).body;
|
||||
await expect.poll(() => starts.mock.calls.filter(([agentId]) => agentId === source.agentId).length, { timeout: 10_000 }).toBe(2);
|
||||
explicitResume = heartbeat.resumeTaskRetry(f.companyId, receipt.requestId);
|
||||
if (mode === "automatic_lost") {
|
||||
await Promise.all([heartbeat.resumeRemoteStopComments(await run(f.runId)),
|
||||
heartbeat.resumeRemoteStopComments(await run(f.runId))]);
|
||||
explicitResume = heartbeat.promoteDueScheduledRetries(new Date(Date.now() + 300_000)).then(() => undefined);
|
||||
} else {
|
||||
await expect.poll(() => starts.mock.calls.filter(([agentId]) => agentId === source.agentId).length, { timeout: 10_000 }).toBe(2);
|
||||
explicitResume = heartbeat.resumeTaskRetry(f.companyId, receipt.requestId);
|
||||
}
|
||||
const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId),
|
||||
sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${receipt.requestId}`));
|
||||
mode === "automatic_lost" ? eq(heartbeatRuns.scheduledRetryReason, "interrupted_preparation")
|
||||
: sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${receipt.requestId}`));
|
||||
expect(successors).toHaveLength(1);
|
||||
if (unavailableCatalog) {
|
||||
const repoUrl = "https://github.com/example/revoked.git";
|
||||
@@ -2226,7 +2255,9 @@ describe("automatic retained workspace recovery", () => {
|
||||
expect((await request(app).get(`/api/issues/${f.issueId}/queued-comments`).expect(200)).body.entries).toEqual([]);
|
||||
expect((await request(app).get(`/api/issues/${f.issueId}`).expect(200)).body.interruption)
|
||||
.toMatchObject({ state: "resolved", canRetry: false });
|
||||
await heartbeat.resumeTaskRetry(f.companyId, receipt.requestId); expect(startTurn).toHaveBeenCalledOnce();
|
||||
if (mode === "automatic_lost") await heartbeat.resumeRemoteStopComments(await run(f.runId));
|
||||
else await heartbeat.resumeTaskRetry(f.companyId, receipt.requestId);
|
||||
expect(startTurn).toHaveBeenCalledOnce();
|
||||
expect((await run(f.runId)).error).toBe(source.error);
|
||||
} finally {
|
||||
try {
|
||||
@@ -2290,6 +2321,216 @@ describe("automatic retained workspace recovery", () => {
|
||||
} finally { await restarted.$client.end(); }
|
||||
}, 30_000);
|
||||
|
||||
it.each([false, true])("continues recovered lost preparation once after its first retry was withheld (startup=%s)", async alreadyRecovered => {
|
||||
const f = await interruptedRemotePreparation("process_lost");
|
||||
const withheld = await withheldPreparationRetry(f);
|
||||
expect(await isPreDispatchExecutionWaitVerified(db, withheld)).toBe(true);
|
||||
if (alreadyRecovered) await recoverRetainedWorkspace(f.input);
|
||||
const restarted = createDb(connectionString);
|
||||
try {
|
||||
const heartbeat = heartbeatService(restarted);
|
||||
await Promise.all(await reconcileRetainedWorkspaces(restarted, f.environmentRuntime,
|
||||
heartbeat.resumeRemoteStopComments, true) ?? []);
|
||||
await Promise.all([heartbeat.resumeRemoteStopComments(await run(f.runId)),
|
||||
heartbeat.resumeRemoteStopComments(await run(f.runId))]);
|
||||
const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId),
|
||||
eq(heartbeatRuns.status, "scheduled_retry")));
|
||||
expect(successors).toHaveLength(1);
|
||||
expect(successors[0]).toMatchObject({ retryOfRunId: withheld.id, scheduledRetryAttempt: 2,
|
||||
scheduledRetryReason: "interrupted_preparation", contextSnapshot: { executionRetryAccounting: { failureRetries: 2 } } });
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
|
||||
expect((await run(f.runId)).status).toBe("failed");
|
||||
expect(await run(withheld.id)).toMatchObject({ status: "cancelled", startedAt: null });
|
||||
expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(true);
|
||||
} finally { await restarted.$client.end(); }
|
||||
}, 30_000);
|
||||
|
||||
it.each(["withheld_first", "successor_first"] as const)("reuses the newest retry child without replacing a withheld sibling (%s)", async order => {
|
||||
const f = await interruptedRemotePreparation("process_lost");
|
||||
const source = await run(f.runId);
|
||||
const now = Date.now();
|
||||
const addSuccessor = async () => (await db.insert(heartbeatRuns).values({
|
||||
companyId: f.companyId, agentId: source.agentId, issueId: f.issueId, retryOfRunId: source.id,
|
||||
status: "failed", runtimeMode: "native", nativeIssueId: f.issueId,
|
||||
startedAt: new Date(now - 1000), finishedAt: new Date(now), createdAt: new Date(now - 1000),
|
||||
errorCode: "native_provider_terminal_failed", contextSnapshot: { issueId: f.issueId },
|
||||
scheduledRetryAttempt: 2, scheduledRetryReason: "transient_failure",
|
||||
}).returning())[0];
|
||||
const addWithheld = async () => {
|
||||
const child = await withheldPreparationRetry(f);
|
||||
await db.update(heartbeatRuns).set({ createdAt: new Date(now - 2000) }).where(eq(heartbeatRuns.id, child.id));
|
||||
return child;
|
||||
};
|
||||
let successor: typeof heartbeatRuns.$inferSelect;
|
||||
let withheld: typeof heartbeatRuns.$inferSelect;
|
||||
if (order === "withheld_first") { withheld = await addWithheld(); successor = await addSuccessor(); }
|
||||
else { successor = await addSuccessor(); withheld = await addWithheld(); }
|
||||
await recoverRetainedWorkspace(f.input);
|
||||
const heartbeat = heartbeatService(db);
|
||||
await Promise.all([heartbeat.resumeRemoteStopComments(await run(f.runId)),
|
||||
heartbeat.resumeRemoteStopComments(await run(f.runId))]);
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, withheld.id))).toHaveLength(0);
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, source.id))).toHaveLength(2);
|
||||
expect(await run(successor!.id)).toEqual(successor!);
|
||||
expect((await action(f.actionId)).evidence).toMatchObject({ bootstrapPreparationContinuation: "delivered",
|
||||
bootstrapPreparationContinuationRunId: successor!.id });
|
||||
expect(await nextWorkspaceRecoveryAt(db)).toBeNull();
|
||||
}, 30_000);
|
||||
|
||||
it.each([false, true])("continues a recovered lost preparation without an earlier retry (startup=%s)", async alreadyRecovered => {
|
||||
const f = await interruptedRemotePreparation("process_lost");
|
||||
if (alreadyRecovered) await recoverRetainedWorkspace(f.input);
|
||||
const restarted = createDb(connectionString);
|
||||
try {
|
||||
const heartbeat = heartbeatService(restarted);
|
||||
await Promise.all(await reconcileRetainedWorkspaces(restarted, f.environmentRuntime,
|
||||
heartbeat.resumeRemoteStopComments, true) ?? []);
|
||||
await Promise.all([heartbeat.resumeRemoteStopComments(await run(f.runId)),
|
||||
heartbeat.resumeRemoteStopComments(await run(f.runId))]);
|
||||
const successors = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, f.runId));
|
||||
expect(successors).toHaveLength(1);
|
||||
expect(successors[0]).toMatchObject({ status: "scheduled_retry", scheduledRetryAttempt: 1 });
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
|
||||
expect(await nextWorkspaceRecoveryAt(restarted)).toBeNull();
|
||||
} finally { await restarted.$client.end(); }
|
||||
}, 30_000);
|
||||
|
||||
it.each(["restart", "pause", "claimed", "duplicate"] as const)("retains preparation continuation after settlement before enqueue (%s)", async mode => {
|
||||
const f = await interruptedRemotePreparation("process_lost"), withheld = await withheldPreparationRetry(f);
|
||||
await recoverRetainedWorkspace(f.input);
|
||||
const source = await run(f.runId);
|
||||
expect(await settleInterruptedNativeBootstrap(db, { run: source, providerDispatchStarted: false })).not.toBeNull();
|
||||
expect((await action(f.actionId)).evidence.automaticRecovery).toMatchObject({ replay: "allowed" });
|
||||
expect(await nextWorkspaceRecoveryAt(db)).not.toBeNull();
|
||||
const duplicateId = mode === "duplicate" ? randomUUID() : null;
|
||||
if (duplicateId) {
|
||||
const prior = await action(f.actionId);
|
||||
await db.insert(issueRecoveryActions).values({ ...prior, id: duplicateId });
|
||||
}
|
||||
if (mode === "claimed") {
|
||||
const first = await heartbeatService(db).scheduleBoundedRetry(f.runId);
|
||||
expect(first.outcome).toBe("scheduled");
|
||||
expect((await action(f.actionId)).evidence.bootstrapPreparationContinuation).toBe("pending");
|
||||
}
|
||||
if (mode === "pause") await db.update(agents).set({ status: "paused" }).where(eq(agents.id, source.agentId));
|
||||
const restarted = createDb(connectionString);
|
||||
try {
|
||||
const heartbeat = heartbeatService(restarted);
|
||||
await Promise.all(await reconcileRetainedWorkspaces(restarted, f.environmentRuntime,
|
||||
heartbeat.resumeRemoteStopComments, true) ?? []);
|
||||
if (mode === "pause") {
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, withheld.id))).toHaveLength(0);
|
||||
expect(await nextWorkspaceRecoveryAt(restarted)).toBeNull();
|
||||
expect((await action(f.actionId)).evidence.bootstrapPreparationContinuation).toBe("deferred");
|
||||
const wake = vi.fn(), unsubscribe = subscribeDeliveryWork(restarted, DELIVERY_QUEUES.workspaceRecovery, wake);
|
||||
try {
|
||||
const rollback = new Error("rollback eligibility change");
|
||||
await expect(restarted.transaction(async tx => {
|
||||
await notifyDeferredExecutionEligibility(tx, { companyId: f.companyId });
|
||||
await tx.update(agents).set({ status: "idle" }).where(eq(agents.id, source.agentId));
|
||||
throw rollback;
|
||||
})).rejects.toBe(rollback);
|
||||
expect((await db.select().from(agents).where(eq(agents.id, source.agentId)))[0].status).toBe("paused");
|
||||
await Promise.all(await reconcileRetainedWorkspaces(restarted, f.environmentRuntime,
|
||||
heartbeat.resumeRemoteStopComments, true) ?? []);
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, withheld.id))).toHaveLength(0);
|
||||
expect(await nextWorkspaceRecoveryAt(restarted)).toBeNull();
|
||||
wake.mockClear();
|
||||
await restarted.transaction(async tx => {
|
||||
await notifyDeferredExecutionEligibility(tx, { companyId: f.companyId });
|
||||
await tx.update(agents).set({ status: "idle" }).where(eq(agents.id, source.agentId));
|
||||
});
|
||||
expect(wake).toHaveBeenCalled();
|
||||
} finally { unsubscribe(); }
|
||||
await Promise.all(await reconcileRetainedWorkspaces(restarted, f.environmentRuntime,
|
||||
heartbeat.resumeRemoteStopComments, true) ?? []);
|
||||
}
|
||||
const successors = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, withheld.id));
|
||||
expect(successors).toHaveLength(1);
|
||||
expect(successors[0]).toMatchObject({ status: "scheduled_retry", scheduledRetryAttempt: 2 });
|
||||
expect((await action(f.actionId)).evidence).toMatchObject({ bootstrapPreparationContinuation: "delivered",
|
||||
bootstrapPreparationContinuationRunId: successors[0].id });
|
||||
expect(await nextWorkspaceRecoveryAt(restarted)).toBeNull();
|
||||
if (duplicateId) expect((await action(duplicateId)).evidence.bootstrapPreparationContinuation).toBe("delivered");
|
||||
expect(await retainedWorkspaceCandidates(restarted, undefined, false, true)).not.toEqual(
|
||||
expect.arrayContaining([expect.objectContaining({ run: expect.objectContaining({ id: f.runId }) })]));
|
||||
} finally { await restarted.$client.end(); }
|
||||
}, 30_000);
|
||||
|
||||
it("revisits a pending continuation after enqueue fails before a new notification", async () => {
|
||||
const f = await interruptedRemotePreparation("process_lost"), withheld = await withheldPreparationRetry(f);
|
||||
await recoverRetainedWorkspace(f.input);
|
||||
expect(await settleInterruptedNativeBootstrap(db, { run: await run(f.runId), providerDispatchStarted: false })).not.toBeNull();
|
||||
const heartbeat = heartbeatService(db), passiveSweep = {};
|
||||
let attempts = 0;
|
||||
const onRecovered = async (source: Awaited<ReturnType<typeof run>>) => {
|
||||
if (source.id !== f.runId) return;
|
||||
if (++attempts === 1) throw new Error("database read unavailable before enqueue notification");
|
||||
await heartbeat.resumeRemoteStopComments(source);
|
||||
};
|
||||
await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, onRecovered, true, undefined, passiveSweep) ?? []);
|
||||
expect(attempts).toBe(1);
|
||||
expect(await nextWorkspaceRecoveryAt(db)).not.toBeNull();
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, withheld.id))).toHaveLength(0);
|
||||
await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, onRecovered, true, undefined, passiveSweep) ?? []);
|
||||
expect(attempts).toBe(2);
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, withheld.id)))
|
||||
.toEqual([expect.objectContaining({ status: "scheduled_retry", scheduledRetryAttempt: 2 })]);
|
||||
expect(await nextWorkspaceRecoveryAt(db)).toBeNull();
|
||||
}, 30_000);
|
||||
|
||||
it.each(["no_certificate", "provider_started", "runtime_selected", "live_controller", "operator_stop", "missing_stop_receipt", "wrong_snapshot", "retry_exhausted", "human_reblock", "paused", "reassigned", "approval"] as const)(
|
||||
"keeps lost preparation behind its %s gate after file preservation", async gate => {
|
||||
const f = await interruptedRemotePreparation("process_lost");
|
||||
await recoverRetainedWorkspace(f.input);
|
||||
const source = await run(f.runId);
|
||||
if (gate === "no_certificate") await db.update(heartbeatRuns).set({ resultJson: { ...source.resultJson,
|
||||
executionRecovery: null } }).where(eq(heartbeatRuns.id, f.runId));
|
||||
if (gate === "provider_started") await db.insert(heartbeatRunEvents).values({ companyId: f.companyId,
|
||||
runId: f.runId, agentId: source.agentId, seq: 1, eventType: "native.process_start_requested", stream: "system" });
|
||||
if (gate === "runtime_selected") await db.update(heartbeatRuns).set({ runtimeModeResolvedAt: new Date() }).where(eq(heartbeatRuns.id, f.runId));
|
||||
if (gate === "live_controller") await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(heartbeatRuns.id, f.runId));
|
||||
if (gate === "operator_stop") await db.update(heartbeatRuns).set({ resultJson: { ...source.resultJson,
|
||||
cancellation: requestedRunCancellation({ cancelledByActorType: "user", cancelledByUserId: "operator" }, "Stop requested"),
|
||||
} }).where(eq(heartbeatRuns.id, f.runId));
|
||||
if (gate === "missing_stop_receipt") {
|
||||
const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id));
|
||||
await db.update(environmentLeases).set({ metadata: { ...lease.metadata, remoteExecutionTermination: null } }).where(eq(environmentLeases.id, f.lease.id));
|
||||
}
|
||||
if (gate === "wrong_snapshot") await db.update(heartbeatRuns).set({ runnerProfileJson: { ...source.runnerProfileJson,
|
||||
remoteWorkspace: { ...(source.runnerProfileJson!.remoteWorkspace as Record<string, unknown>), snapshotId: randomUUID() },
|
||||
} }).where(eq(heartbeatRuns.id, f.runId));
|
||||
if (gate === "retry_exhausted") await db.update(heartbeatRuns).set({ scheduledRetryAttempt: 2 }).where(eq(heartbeatRuns.id, f.runId));
|
||||
if (gate === "human_reblock") await db.update(issues).set({ statusVersion: 3 }).where(eq(issues.id, f.issueId));
|
||||
if (gate === "paused") await db.update(agents).set({ status: "paused" }).where(eq(agents.id, source.agentId));
|
||||
if (gate === "reassigned") await db.update(issues).set({ assigneeAgentId: null, assigneeUserId: "board" }).where(eq(issues.id, f.issueId));
|
||||
if (gate === "approval") {
|
||||
const approvalId = randomUUID();
|
||||
await db.insert(approvals).values({ id: approvalId, companyId: f.companyId, type: "hire_agent", status: "pending", payload: {} });
|
||||
await db.insert(issueApprovals).values({ companyId: f.companyId, issueId: f.issueId, approvalId });
|
||||
}
|
||||
const heartbeat = heartbeatService(db);
|
||||
await heartbeat.resumeRemoteStopComments(await run(f.runId));
|
||||
expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId),
|
||||
eq(heartbeatRuns.retryOfRunId, f.runId)))).toHaveLength(0);
|
||||
expect((await run(f.runId)).error).toBe(source.error);
|
||||
}, 30_000);
|
||||
|
||||
it.each(["provider_evidence", "exhausted"] as const)("does not replace a withheld retry with %s", async gate => {
|
||||
const f = await interruptedRemotePreparation("process_lost"), withheld = await withheldPreparationRetry(f);
|
||||
if (gate === "provider_evidence") await db.insert(heartbeatRunEvents).values({ companyId: f.companyId,
|
||||
runId: withheld.id, agentId: withheld.agentId, seq: 1, eventType: "tool.execution.started", stream: "system" });
|
||||
if (gate === "exhausted") await db.update(heartbeatRuns).set({ scheduledRetryAttempt: 2 }).where(eq(heartbeatRuns.id, withheld.id));
|
||||
await recoverRetainedWorkspace(f.input);
|
||||
const heartbeat = heartbeatService(db);
|
||||
await Promise.all([heartbeat.resumeRemoteStopComments(await run(f.runId)),
|
||||
heartbeat.resumeRemoteStopComments(await run(f.runId))]);
|
||||
expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId),
|
||||
eq(heartbeatRuns.retryOfRunId, withheld.id)))).toHaveLength(0);
|
||||
expect(await run(withheld.id)).toMatchObject({ status: "cancelled", startedAt: null });
|
||||
expect(await readFile(path.join(f.remote, "saved.txt"), "utf8")).toBe("finished work\n");
|
||||
}, 30_000);
|
||||
|
||||
it("retries unpublished preparation from its preserved ancestor without an old failure-block receipt", async () => {
|
||||
const f = await seed("legacy", false);
|
||||
await requestWorkspaceRecovery({ ...f, idempotencyKey: randomUUID(), actorId: "board" });
|
||||
|
||||
@@ -496,10 +496,16 @@ export function retainedWorkspaceCandidates(db: Db, cursor?: string, unfinishedO
|
||||
and ${agentWakeupRequests.status} = 'deferred_issue_execution' and ${agentWakeupRequests.requestedByActorType} = 'user'
|
||||
and ${agentWakeupRequests.payload}->>'issueId' = ${issueRecoveryActions.sourceIssueId}::text)`;
|
||||
const interruptedPreparation = sql`(${heartbeatRuns.runtimeMode} = 'legacy'
|
||||
and ${heartbeatRuns.status} = 'interrupted'
|
||||
and (${heartbeatRuns.status} = 'interrupted' or (${heartbeatRuns.status} = 'failed'
|
||||
and ${heartbeatRuns.errorCode} = 'process_lost'
|
||||
and ${heartbeatRuns.resultJson}#>>'{executionRecovery,kind}' = 'bootstrap'
|
||||
and ${heartbeatRuns.resultJson}#>'{executionRecovery,providerWorkStarted}' = 'false'::jsonb))
|
||||
and ${heartbeatRuns.runtimeModeResolvedAt} is null and ${heartbeatRuns.executionStage} = 'preparing'
|
||||
and ${heartbeatRuns.runnerProfileJson}#>>'{adapterDispatch,adapterType}' = 'paperclip_runner'
|
||||
and ${issueRecoveryActions.evidence}#>>'{automaticRecovery,replay}' = 'blocked'
|
||||
and (${issueRecoveryActions.evidence}#>>'{automaticRecovery,replay}' = 'blocked'
|
||||
or (${issueRecoveryActions.evidence}#>>'{automaticRecovery,replay}' = 'allowed'
|
||||
and ${issueRecoveryActions.evidence}->>'bootstrapPreparationSettledAt' is not null
|
||||
and coalesce(${issueRecoveryActions.evidence}->>'bootstrapPreparationContinuation', 'pending') in ('pending', 'deferred')))
|
||||
and exists (select 1 from ${issues} where ${issues.companyId} = ${heartbeatRuns.companyId}
|
||||
and ${issues.id} = ${issueRecoveryActions.sourceIssueId} and ${issues.assigneeAgentId} = ${heartbeatRuns.agentId}
|
||||
and ${issues.status} in ('todo', 'in_progress', 'in_review', 'blocked')
|
||||
@@ -512,7 +518,10 @@ export function retainedWorkspaceCandidates(db: Db, cursor?: string, unfinishedO
|
||||
then (${issueRecoveryActions.evidence}->>'runId')::uuid end`)))
|
||||
.where(and(inArray(heartbeatRuns.status, terminal),
|
||||
workspaceRecoverySourceEligible(true),
|
||||
unfinishedOnly ? sql`${issueRecoveryActions.evidence}#>>'{systemRecovery,state}' in ('scheduled', 'running', 'awaiting_evidence')` : undefined,
|
||||
unfinishedOnly ? sql`(${issueRecoveryActions.evidence}#>>'{systemRecovery,state}' in ('scheduled', 'running', 'awaiting_evidence')
|
||||
or (${issueRecoveryActions.evidence}#>>'{systemRecovery,state}' = 'complete'
|
||||
and ${issueRecoveryActions.evidence}->>'bootstrapPreparationSettledAt' is not null
|
||||
and coalesce(${issueRecoveryActions.evidence}->>'bootstrapPreparationContinuation', 'pending') = 'pending'))` : undefined,
|
||||
obligationsOnly ? sql`(${issueRecoveryActions.evidence}#>>'{systemRecovery,state}' in ('scheduled', 'running', 'awaiting_evidence')
|
||||
or (${issueRecoveryActions.evidence}#>>'{systemRecovery,state}' = 'complete' and (${cleanup} or ${interruptedPreparation})) or ${waitingInput})` : undefined,
|
||||
sql`(${issueRecoveryActions.evidence}#>>'{systemRecovery,state}' in ('scheduled', 'running', 'awaiting_evidence', 'complete', 'exhausted')
|
||||
|
||||
Reference in new issue
Block a user