diff --git a/server/src/services/cancelled-native-startup.ts b/server/src/services/cancelled-native-startup.ts index f7e13b561c..9a26fa249a 100644 --- a/server/src/services/cancelled-native-startup.ts +++ b/server/src/services/cancelled-native-startup.ts @@ -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 { + 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 { return hasSettledComputerAdmissionPreparation(run) && isComputerAdmissionWaitBeforeProvider(db, run); diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index 8240f50932..02a62ee6a1 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -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. diff --git a/server/src/services/heartbeat/run-control.test.ts b/server/src/services/heartbeat/run-control.test.ts index 7898b9f706..a06b7501b8 100644 --- a/server/src/services/heartbeat/run-control.test.ts +++ b/server/src/services/heartbeat/run-control.test.ts @@ -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)) { diff --git a/server/src/services/heartbeat/run-control.ts b/server/src/services/heartbeat/run-control.ts index beebcafc1e..1aeb7943c8 100644 --- a/server/src/services/heartbeat/run-control.ts +++ b/server/src/services/heartbeat/run-control.ts @@ -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; } diff --git a/server/src/services/legacy-execution-recovery.ts b/server/src/services/legacy-execution-recovery.ts index fc874bd90b..4929e24aab 100644 --- a/server/src/services/legacy-execution-recovery.ts +++ b/server/src/services/legacy-execution-recovery.ts @@ -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 { 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. */