fix: resume new input after paused computer admission retries

This commit is contained in:
Dotta committed 2026-10-10 22:48:08 -05:00
1 parent 4baff4f537
commit a5cc05d2c3
5 files changed
+154 -9

No files matched your search

@@ -1,7 +1,7 @@
import { isComputerAdmissionWait } from "./execution-recovery-attempt.js";
import { adapterExecutionControls } from "./adapter-execution-control.js";
import { isPreDispatchReviewWait } from "./pre-dispatch-review-wait.js";
import { and, eq, inArray, isNotNull, or } from "drizzle-orm";
import { and, eq, inArray, isNotNull, or, sql } from "drizzle-orm";
import { computers, environmentLeases, heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db";
import { claimedAdapterType } from "./conversation-continuation.js";
import { PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED } from "./native-local-process-stop.js";
@@ -113,6 +113,49 @@ export function hasSettledComputerAdmissionPreparation(run: Run): boolean {
return true;
}
/** A paused scheduled computer retry never reached a preparing controller.
* This is evidence for new user input, never automatic replay of the old wake. */
export async function isPausedComputerAdmissionRetryBeforeProvider(db: Db, run: Run): Promise<boolean> {
const issueId = run.contextSnapshot?.issueId;
if (run.status !== "cancelled" || run.errorCode !== "issue_paused" ||
run.scheduledRetryReason !== "computer_admission_wait" || !run.retryOfRunId ||
!run.scheduledRetryAt || !run.finishedAt || typeof issueId !== "string" ||
run.startedAt || run.runtimeMode !== "legacy" || run.runtimeModeResolvedAt || run.executionStage ||
run.controllerBootId || run.controllerLeaseExpiresAt || run.nativeIssueId || run.nativeSessionId ||
run.runnerInstanceId || run.processPid || run.processGroupId || run.processStartedAt || run.sessionIdAfter ||
adapterExecutionControls.has(run.id)) return false;
const [parent] = await db.select().from(heartbeatRuns).where(and(
eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.id, run.retryOfRunId),
eq(heartbeatRuns.agentId, run.agentId), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`,
)).limit(1);
if (!parent || !await canRetryComputerAdmissionWait(db, parent)) return false;
const [receipt] = await db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and(
eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id),
eq(heartbeatRunEvents.agentId, run.agentId), eq(heartbeatRunEvents.eventType, "lifecycle"),
eq(heartbeatRunEvents.stream, "system"),
eq(heartbeatRunEvents.message, "Scheduled retry suppressed because the issue is held by an active subtree pause hold"),
sql`${heartbeatRunEvents.payload}->>'issueId' = ${issueId}`,
sql`${heartbeatRunEvents.payload}->>'scheduledRetryReason' = 'computer_admission_wait'`,
sql`${heartbeatRunEvents.payload}->>'scheduledRetryAt' = ${run.scheduledRetryAt.toISOString()}`,
sql`${heartbeatRunEvents.payload}->>'holdId' is not null`,
)).limit(1);
if (!receipt) return false;
const [lease] = await db.select({ id: environmentLeases.id }).from(environmentLeases).where(and(
eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id),
)).limit(1);
const [coordinator] = await db.select({ id: nativeRunFinalizations.runId }).from(nativeRunFinalizations).where(and(
eq(nativeRunFinalizations.companyId, run.companyId), eq(nativeRunFinalizations.runId, run.id),
)).limit(1);
const [execution] = await db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and(
eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id),
or(isNotNull(heartbeatRunEvents.sourceEventId),
inArray(heartbeatRunEvents.eventType, ["adapter.invoke", PROCESS_START_REQUESTED, PROCESS_IDENTITY_RECORDED,
"harness.ready", "session.started", "session.resumed", "session.updated", "turn.started",
"provider.event", "provider.rpc_result", "tool.execution.started"])),
)).limit(1);
return !lease && !coordinator && !execution;
}
/** Scheduling additionally requires cleanup or an expired prior controller. */
export async function canRetryComputerAdmissionWait(db: Db, run: Run): Promise<boolean> {
return hasSettledComputerAdmissionPreparation(run) && isComputerAdmissionWaitBeforeProvider(db, run);
@@ -1,4 +1,4 @@
import { canRetryComputerAdmissionWait } from "./cancelled-native-startup.js";
import { canRetryComputerAdmissionWait, isPausedComputerAdmissionRetryBeforeProvider } from "./cancelled-native-startup.js";
import { hasRequiredWorkspaceRecovery } from "./workspace-restore-recovery-state.js";
import { createHash } from "node:crypto";
import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js";
@@ -204,7 +204,10 @@ export async function admitExplicitNativeContinuation(input: {
if (run.resultJson?.workspaceRestoreFailure === "restore_unsafe_archive" || hasRequiredWorkspaceRecovery(run.resultJson)) {
return blocked("workspace_repair_required", "Verify safe workspace staging or repair before continuing. Your message is saved.");
}
const cancelledStartup = await isCancelledNativeStartup(db, run, coordinator);
const pausedComputerRetry = action.cause === "legacy_execution_requires_reconciliation" &&
action.fingerprint === `legacy-execution:${run.id}` && !retry && !queuedInterrupt && !input.queuedCommentInterruptId &&
authorizedAt > run.finishedAt! && await isPausedComputerAdmissionRetryBeforeProvider(db, run);
const cancelledStartup = pausedComputerRetry || await isCancelledNativeStartup(db, run, coordinator);
if (partiallyDeliveredQueue && run.runtimeMode !== "native" && !unusedAdmission && !cancelledStartup) return null;
if ((queuedInterrupt || queuedRequest) && !legacyUserTurn && !unusedAdmission &&
!(queuedRequest && run.runtimeMode === "native" &&
@@ -218,7 +221,7 @@ export async function admitExplicitNativeContinuation(input: {
!(cancelledStartup && authorizedAt > run.finishedAt!)) return null;
if (retry && run.status === "cancelled" && !canContinueCancelledRun(run) && !cancelledStartup)
return blocked("cancelled_by_operator", "Inspect the stopped run and send a new message to continue.");
if (cancelledStartup) cancelledStartupIds.add(run.id);
if (cancelledStartup && !pausedComputerRetry) cancelledStartupIds.add(run.id);
if (run.runtimeMode !== "native" && !unusedAdmission && !legacyUserTurn && !cancelledStartup) return null;
// A provider failure can finish the normal result/assessment commit path.
// Its accepted failed result is immutable history, not a live controller.
@@ -1,3 +1,6 @@
import { admitExplicitNativeContinuation } from "../explicit-native-continuation.js";
import { legacyExecutionNeedsReconciliationWithEvidence, terminalizeLegacyExecution } from "../legacy-execution-recovery.js";
import { createPostgresRunDispatchAdapter } from "../../modules/run-dispatch/adapters/postgres.js";
import { createHeartbeatRetries } from "./retries.js";
import { createRunDispatch } from "../../modules/run-dispatch/index.js";
import { ComputerStopPendingError } from "../../modules/computers/index.js";
@@ -5,7 +8,7 @@ import { heartbeatService } from "../heartbeat.js";
import { canRetryComputerAdmissionWait } from "../cancelled-native-startup.js";
import { randomUUID } from "node:crypto";
import type { Db } from "@paperclipai/db";
import { issueComments, issueRecoveryActions, computers, environments, agents, agentWakeupRequests, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issues, projects } from "@paperclipai/db";
import { environmentLeases, heartbeatRunEvents, issueTreeHolds, issueComments, issueRecoveryActions, computers, environments, agents, agentWakeupRequests, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issues, projects } from "@paperclipai/db";
import { eq, sql } from "drizzle-orm";
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
@@ -203,6 +206,100 @@ describe.skipIf(!support.supported)("heartbeat run-control database wiring", ()
await db.update(issues).set({ executionRunId: run.id }).where(eq(issues.id, issue.id));
return { ...f, issue, run };
}
async function pausedComputerRetryFixture() {
const f = await computerWaitFixture();
await db.update(agents).set({ adapterType: "paperclip_runner", runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } } }).where(eq(agents.id, f.agent.id));
const scheduled = await heartbeatService(db).scheduleBoundedRetry(f.run.id, { retryReason: "computer_admission_wait", delayMs: 0 });
if (scheduled.outcome !== "scheduled") throw new Error("Expected computer wait successor");
const [hold] = await db.insert(issueTreeHolds).values({ companyId: f.company.id, rootIssueId: f.issue.id,
mode: "pause", status: "active", reason: "Pause test", releasePolicy: { strategy: "manual" } }).returning();
expect(await createPostgresRunDispatchAdapter(db).promoteOrCancelDueRetry({ runId: scheduled.run.id,
companyId: f.company.id, now: new Date(Date.now() + 60_000) })).toMatchObject({ outcome: "gate_suppressed", errorCode: "issue_paused" });
const [suppressed] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, scheduled.run.id));
await db.update(issueTreeHolds).set({ status: "released", releasedAt: new Date() }).where(eq(issueTreeHolds.id, hold.id));
return { ...f, suppressed };
}
it("pause suppresses an unstarted computer retry without inventing recovery, and a new message after resume queues once", async () => {
const f = await pausedComputerRetryFixture();
expect(await legacyExecutionNeedsReconciliationWithEvidence(db, f.suppressed)).toBe(false);
await terminalizeLegacyExecution({ db, run: f.suppressed, status: "cancelled", reconcileIfNeeded: true });
expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issue.id))).toHaveLength(0);
// Keep the executor out of this integration test while exercising real wake admission.
await db.insert(heartbeatRuns).values({ companyId: f.company.id, agentId: f.agent.id, status: "running" });
const [comment] = await db.insert(issueComments).values({ companyId: f.company.id, issueId: f.issue.id,
authorType: "user", authorUserId: "board", body: "Read the saved file after resume." }).returning();
const service = heartbeatService(db);
const wake = { source: "automation" as const, triggerDetail: "system" as const, reason: "issue_commented",
requestedByActorType: "user" as const, requestedByActorId: "board", idempotencyKey: `paused-computer:${comment.id}`,
payload: { issueId: f.issue.id, commentId: comment.id }, contextSnapshot: { issueId: f.issue.id, wakeCommentId: comment.id } };
await service.wakeup(f.agent.id, wake);
await service.wakeup(f.agent.id, wake);
const queued = await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.contextSnapshot}->>'wakeCommentId' = ${comment.id}`);
expect(queued).toHaveLength(1);
expect(queued[0]).toMatchObject({ status: "queued", agentId: f.agent.id });
expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issue.id))).toHaveLength(0);
});
it.each(["valid", "saved-message", "saved-old-message", "foreign-parent", "started", "lease", "provider-event", "missing-receipt", "changed-parent", "retry", "interrupt", "deleted-comment", "old-comment"])("historical paused computer retry accepts only verified new-message authority (%s)", async kind => {
const f = await pausedComputerRetryFixture();
if (kind === "foreign-parent") {
const [other] = await db.insert(agents).values({ companyId: f.company.id, name: "Foreign parent agent", adapterType: "paperclip_runner" }).returning();
await db.update(heartbeatRuns).set({ agentId: other.id }).where(eq(heartbeatRuns.id, f.run.id));
}
if (kind === "started") await db.update(heartbeatRuns).set({ startedAt: new Date() }).where(eq(heartbeatRuns.id, f.suppressed.id));
if (kind === "lease") await db.insert(environmentLeases).values({ companyId: f.company.id, issueId: f.issue.id, heartbeatRunId: f.suppressed.id, status: "released", releasedAt: new Date(), cleanupStatus: "success" });
if (kind === "provider-event") await db.insert(heartbeatRunEvents).values({ companyId: f.company.id, agentId: f.agent.id, runId: f.suppressed.id, seq: 999, eventType: "turn.started" });
if (kind === "missing-receipt") await db.delete(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.suppressed.id));
if (kind === "changed-parent") await db.update(heartbeatRuns).set({ resultJson: {} }).where(eq(heartbeatRuns.id, f.run.id));
const [action] = await db.insert(issueRecoveryActions).values({ companyId: f.company.id, sourceIssueId: f.issue.id,
kind: "active_run_watchdog", cause: "legacy_execution_requires_reconciliation", fingerprint: `legacy-execution:${f.suppressed.id}`,
status: "resolved", outcome: "blocked", ownerType: "board", nextAction: "Historical mistaken hold",
evidence: { runId: f.suppressed.id, automaticRecovery: { policy: "preserve_without_replay_v1", replay: "blocked" } } }).returning();
const [comment] = await db.insert(issueComments).values({ companyId: f.company.id, issueId: f.issue.id,
authorType: "user", authorUserId: "board", body: "A new instruction", createdAt: new Date(f.suppressed.finishedAt!.getTime() + (["old-comment", "saved-old-message"].includes(kind) ? -1000 : 1000)),
...(kind === "deleted-comment" ? { deletedAt: new Date() } : {}) }).returning();
if (kind.startsWith("saved-")) {
await db.insert(heartbeatRuns).values({ companyId: f.company.id, agentId: f.agent.id, status: "running" });
const [saved] = await db.insert(agentWakeupRequests).values({ companyId: f.company.id, agentId: f.agent.id,
source: "automation", reason: "issue_commented", status: "deferred_issue_execution",
requestedByActorType: "user", requestedByActorId: "board", updatedAt: new Date(Date.now() - 60_000),
payload: { issueId: f.issue.id, commentId: comment.id, executionWait: { reason: "execution_recovery", recoveryActionId: action.id },
_paperclipWakeContext: { issueId: f.issue.id, wakeReason: "issue_commented", wakeCommentId: comment.id, wakeCommentIds: [comment.id] } },
}).returning();
const service = heartbeatService(db);
await service.resumeExecutionWaitComments();
await service.resumeExecutionWaitComments();
const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, saved.id));
const turns = await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.contextSnapshot}->>'wakeCommentId' = ${comment.id}`);
if (kind === "saved-message") {
expect(wake.status).toBe("coalesced");
expect(turns).toHaveLength(1);
expect(turns[0]).toMatchObject({ status: "queued" });
expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, action.id)))[0].evidence.explicitUserContinuation).toMatchObject({ commentId: comment.id });
} else {
expect(wake.status).toBe("deferred_issue_execution");
expect(turns).toHaveLength(0);
}
return;
}
const result = await db.transaction(async tx => {
await tx.select().from(issues).where(eq(issues.id, f.issue.id)).for("update");
return admitExplicitNativeContinuation({ db: tx as unknown as Db, companyId: f.company.id, issueId: f.issue.id, agentId: f.agent.id,
actorType: "user", actorId: "board", reason: kind === "retry" ? "retry_failed_run" : "issue_commented", commentId: comment.id,
successorRunId: randomUUID(), ...(kind === "retry" ? { failedRunId: f.suppressed.id } : {}),
...(kind === "interrupt" ? { queuedCommentInterruptId: randomUUID() } : {}) });
});
if (kind === "valid") {
expect(result).toMatchObject({ previousRunId: f.suppressed.id, commentId: comment.id });
const [updated] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, action.id));
expect(updated.evidence.explicitUserContinuation).toMatchObject({ commentId: comment.id, previousRunId: f.suppressed.id });
} else {
expect(result).toBeNull();
expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, action.id)))[0].evidence.automaticRecovery).toMatchObject({ replay: "blocked" });
}
});
const operatorStop = { resultJson: { cancelledByActorType: "user", cancelledByUserId: "board" } };
function lifecycleRetries(deps = callbacks(db)) {
+4 -2
View File
@@ -31,7 +31,7 @@ import {
admitExplicitNativeContinuation,
undeliveredLegacyUserCommentIds,
} from "../explicit-native-continuation.js";
import { isCancelledNativeStartup, isComputerAdmissionWaitBeforeProvider, hasSettledComputerAdmissionPreparation } from "../cancelled-native-startup.js";
import { isPausedComputerAdmissionRetryBeforeProvider, isCancelledNativeStartup, isComputerAdmissionWaitBeforeProvider, hasSettledComputerAdmissionPreparation } from "../cancelled-native-startup.js";
import {
executionBlockerPredicate,
getExecutionBlocker,
@@ -386,7 +386,8 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont
? await db.select().from(nativeRunFinalizations).where(and(
eq(nativeRunFinalizations.companyId, run.companyId), eq(nativeRunFinalizations.runId, run.id),
)) : [];
const cancelledPreparation = run.runtimeMode === "legacy" &&
const pausedComputerRetry = await isPausedComputerAdmissionRetryBeforeProvider(db, run);
const cancelledPreparation = pausedComputerRetry || run.runtimeMode === "legacy" &&
await isCancelledNativeStartup(db, run, preparationCoordinator);
if (run.runtimeMode !== "native" && !cancelledPreparation &&
parseObject(run.resultJson?.startupCancellation).beforeNativeSelection !== true &&
@@ -422,6 +423,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont
if (wake.idempotencyKey?.startsWith("chat-inbound:")) continue;
let payload = parseObject(wake.payload);
if (payload.queuedCommentInterrupt) {
if (pausedComputerRetry) continue;
await resumeQueuedCommentInterrupt(wake.companyId, wake.id);
continue;
}
@@ -1,4 +1,4 @@
import { isComputerAdmissionWaitBeforeProvider } from "./cancelled-native-startup.js";
import { isComputerAdmissionWaitBeforeProvider, isPausedComputerAdmissionRetryBeforeProvider } from "./cancelled-native-startup.js";
import { hasRequiredWorkspaceRecovery, preserveWorkspaceRestoreRecoveryMetadata, LEGACY_WORKSPACE_RECOVERY_SCHEMA } from "./workspace-restore-recovery-state.js";
import { hasUnrestoredRemoteWorkspace, preserveLegacyWorkspaceRestoreSources, type LegacyWorkspaceRestoreSource } from "./legacy-workspace-restore-recovery.js";
import { isPreDispatchReviewWaitVerified } from "./pre-dispatch-review-wait.js";
@@ -62,7 +62,7 @@ export function legacyExecutionNeedsReconciliation(
export async function legacyExecutionNeedsReconciliationWithEvidence(db: Db, run: Run): Promise<boolean> {
return await hasUnrestoredRemoteWorkspace(db, run) ||
(legacyExecutionNeedsReconciliation(run) && !(await isPreDispatchReviewWaitVerified(db, run)) &&
!(await isComputerAdmissionWaitBeforeProvider(db, run)));
!(await isComputerAdmissionWaitBeforeProvider(db, run)) && !(await isPausedComputerAdmissionRetryBeforeProvider(db, run)));
}
/** Persist the failed legacy run, owned lock release and operator decision together. */