diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 91a64f1acb5..e09991a6209 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1249,6 +1249,19 @@ Newer saved input keeps its own receipt while the existing Retry explanation sta visible. Terminal waits direct the user to send new input and hide the Retry action that cannot continue that attempt. +Dispatch never declines a board Retry silently. When the intent itself can no +longer start, the intent keeps a named wait and the task shows it. If its saved +board message changed (`retry_source_changed`), the wait is terminal: Retry is +hidden and the user sends a new message. If its requester can no longer change +the task (`retry_authority_denied`), Retry stays available: a later click with a +new idempotency key replaces that intent with a fresh attempt under the clicker's current authority, instead of joining it, and starts +at most one run. The replaced intent is recorded as `skipped` with +`task_retry_replaced`. Clicks that arrive while a Retry waits on shared state +(cleanup, recovery, an active run, preservation) still join that Retry; it +starts by itself once the gate clears. A Retry accepted during the pre-deploy +drain stays saved and is dispatched by the deferred-receipt worker after the +restart. + A remote native run can lose its original runner or sandbox before reattachment. This happens after a server restart, and for a same-run retry after a turn ends without a semantic result. The reattachment then fails with diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index df1fc858f91..7274724bb7d 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -2828,13 +2828,23 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende if (pending.id !== opts.taskRetryRequestId || retry?.generationId !== pending.id || retry.chatActionId || retry.actorId !== opts.requestedByActorId || retry.sourceRunId !== enrichedContextSnapshot.previousRunId) return { kind: "deferred" as const }; + // A declined Retry stays saved, but never silently: name the gate + // so the task explains it and a new click can replace this intent. + const declineRetry = async (reason: string, message: string) => { + await writeDeferredExecutionWait(tx as unknown as Db, { companyId: agent.companyId, + requestId: pending.id }, { reason, message }, { replace: true }); + logger.warn({ requestId: pending.id, issueId, reason }, "Task Retry could not start"); + return { kind: "deferred" as const }; + }; if (!await taskRetryMutationAllowed(tx as unknown as Db, { companyId: agent.companyId, issueId, - actorId: retry.actorId, actorSource: retry.actorSource })) return { kind: "deferred" as const }; + actorId: retry.actorId, actorSource: retry.actorSource })) + return declineRetry("retry_authority_denied", "The user who requested Retry can no longer change this task."); if (retry.boardMessageOrigin) { const origin = await readTaskRetryBoardMessageOrigin(tx as unknown as Db, { companyId: agent.companyId, issueId, agentId, sourceRunId: retry.sourceRunId, actorId: retry.actorId }); if (!origin || origin.commentId !== retry.boardMessageOrigin.commentId || - origin.identityContextId !== retry.boardMessageOrigin.identityContextId) return { kind: "deferred" as const }; + origin.identityContextId !== retry.boardMessageOrigin.identityContextId) + return declineRetry("retry_source_changed", "The saved message this Retry continues has changed."); } } if (opts.queuedCommentRequestId) { diff --git a/server/src/services/task-interruption-presentation.test.ts b/server/src/services/task-interruption-presentation.test.ts index 749e952a807..adbc79479a5 100644 --- a/server/src/services/task-interruption-presentation.test.ts +++ b/server/src/services/task-interruption-presentation.test.ts @@ -95,13 +95,20 @@ it("explains a persisted Retry wait without exposing the saved payload", () => { .toMatchObject({ state: "waiting_for_user", userAction: "resume" }); }); -it.each(["cleanup_quarantined", "cancelled_by_operator"])("does not offer Retry for terminal Retry waits: %s", reason => { +it.each(["cleanup_quarantined", "cancelled_by_operator", "retry_source_changed"])("does not offer Retry for terminal Retry waits: %s", reason => { const failed = { ...run, errorCode: "native_provider_terminal_failed", resultJson: null }; expect(projectTaskInterruption(issue, { run: failed, agent, paused: false, approval: false, wait: { id: "retry", requestedAt: new Date(), sourceRunId: failed.id, executionWaitReason: reason } })) .toMatchObject({ state: "failed", canRetry: false, retryWait: { reason } }); }); +it("keeps Retry for a Retry declined for its requester's access, since a new click replaces it", () => { + const failed = { ...run, errorCode: "native_provider_terminal_failed", resultJson: null }; + expect(projectTaskInterruption(issue, { run: failed, agent, paused: false, approval: false, + wait: { id: "retry", requestedAt: new Date(), sourceRunId: failed.id, executionWaitReason: "retry_authority_denied" } })) + .toMatchObject({ state: "failed", canRetry: true, retryWait: { reason: "retry_authority_denied" } }); +}); + it("preserves the actual workspace preparation error without changing retry authority", () => { const failedRun = { ...run, errorCode: "workspace_validation_failed", error: "The selected project has no repository source for a remote workspace.", resultJson: null }; diff --git a/server/src/services/task-interruption-presentation.ts b/server/src/services/task-interruption-presentation.ts index 4a2e5de19b7..59797d3958e 100644 --- a/server/src/services/task-interruption-presentation.ts +++ b/server/src/services/task-interruption-presentation.ts @@ -68,6 +68,11 @@ const retryWaitMessages: Record = { finalizing_saved_result: "Paperclip is finishing the saved result before Retry can continue.", cleanup_quarantined: "Retry cannot continue this attempt. Send a new message after the previous execution has stopped.", cancelled_by_operator: "This attempt was stopped. Send a new message to continue from saved work.", + // A new click replaces a Retry declined for its requester's access with a + // fresh attempt under the clicker's access, so Retry stays available. + retry_authority_denied: "Retry could not start with the requesting user's access. Press Retry to start it with yours.", + // An edited saved message can never be replayed; only new input continues. + retry_source_changed: "Retry cannot continue because its saved message changed. Send a new message to continue from saved work.", }; /** Recognize old persisted failures without displaying arbitrary run error text. */ @@ -152,7 +157,7 @@ export function projectTaskInterruption(issue: ProjectionIssue, input: Projectio const reason = Object.hasOwn(retryWaitMessages, retryWait.executionWaitReason) ? retryWait.executionWaitReason : "execution_recovery"; // Keep the latest input receipt so message submission can settle independently. return { ...base, state: "failed", - canRetry: !["cleanup_quarantined", "cancelled_by_operator"].includes(reason), retryWait: { reason, + canRetry: !["cleanup_quarantined", "cancelled_by_operator", "retry_source_changed"].includes(reason), retryWait: { reason, message: retryWaitMessages[reason] ?? "Retry is waiting for execution checks. Your message is saved." } }; } if (wait && wait.requestedAt.getTime() > now - 10_000 && system?.state !== "exhausted") diff --git a/server/src/services/task-retry.ts b/server/src/services/task-retry.ts index 0ac8126239f..45c347455ca 100644 --- a/server/src/services/task-retry.ts +++ b/server/src/services/task-retry.ts @@ -56,6 +56,17 @@ const taskRetryIntentSchema = z.object({ boardMessageOrigin: z.object({ commentId: z.string().uuid(), identityContextId: z.string().uuid() }).optional(), }); export type TaskRetryIntent = z.infer; + +/** Dispatch declines a Retry intent for this reason when its own requester, + * not the task's shared state, prevents it from starting; a click by a user + * with access replaces it. An edited saved board message + * (`retry_source_changed`) is terminal: only new input continues. */ +export const DECLINED_TASK_RETRY_REASONS = ["retry_authority_denied"] as const; +export function isDeclinedTaskRetry(payload: Record | null | undefined) { + const wait = payload?.executionWait; + const reason = wait && typeof wait === "object" && !Array.isArray(wait) ? (wait as Record).reason : null; + return DECLINED_TASK_RETRY_REASONS.some(value => value === reason); +} export function readTaskRetryIntent(payload: Record | null | undefined): TaskRetryIntent | null { const result = taskRetryIntentSchema.safeParse(payload?.taskRetry); return result.success ? result.data : null; @@ -315,6 +326,22 @@ function admitTaskRetry(input: TaskRetryRequest) { .where(eq(agentWakeupRequests.id, active.id)); active = undefined; } + // Dispatch declined this unadmitted intent for a reason bound to the intent + // itself (its requester's access or saved source), so it can never start. + // A new deliberate click (new key) replaces it with a fresh attempt under the + // clicker's current authority instead of joining it. Waits on shared state + // (cleanup, recovery, an active run) keep coalescing: that intent starts by + // itself once the gate clears. The restart hook never replaces a user's click. + let replacedRequestId: string | undefined; + if (active && !busy && !active.runId && active.status === "deferred_issue_execution" && + input.initiatedBy !== "restart_recovery" && activeIntent && !activeIntent.chatActionId && + !activeIntent.queuedChatRequestId && isDeclinedTaskRetry(active.payload)) { + await tx.update(agentWakeupRequests).set({ status: "skipped", error: "task_retry_replaced", + finishedAt: new Date(), updatedAt: new Date() }).where(and(eq(agentWakeupRequests.id, active.id), + eq(agentWakeupRequests.status, "deferred_issue_execution"), sql`${agentWakeupRequests.runId} is null`)); + replacedRequestId = active.id; + active = undefined; + } const id = randomUUID(); const intent: TaskRetryIntent = { schema: "paperclip.task-retry.v1", generationId: active?.id ?? id, sourceRunId: latest.id, actorId: input.actorId, @@ -400,6 +427,7 @@ function admitTaskRetry(input: TaskRetryRequest) { actorId: restartRecovery ? "restart-recovery" : input.actorId, action: "issue.retry_requested", entityType: "issue", entityId: issue.id, runId: latest.id, details: { requestId: active?.id ?? id, sourceRunId: latest.id, attachedToExistingWork: Boolean(active || busy), + ...(replacedRequestId ? { replacedRequestId } : {}), ...(restartRecovery ? { initiatedBy: "restart_recovery", onBehalfOfUserId: input.actorId } : {}) } }); return { requestId: active?.id ?? id, accepted: true, stale: false, observedRunId: latest.id, chatActionId: intent.chatActionId }; diff --git a/server/src/services/workspace-recovery.test.ts b/server/src/services/workspace-recovery.test.ts index f6e82ce4404..a8f766039d8 100644 --- a/server/src/services/workspace-recovery.test.ts +++ b/server/src/services/workspace-recovery.test.ts @@ -43,6 +43,7 @@ import { adoptHistoricalWorkspaceRecoveries, historicalRecoveryCounts } from "./ import { createSystemRecovery, failSystemRecovery, reserveSystemRecoveryAttempt, SystemRecoveryFailure } from "./system-recovery-state.js"; import { requestTaskRetry } from "./task-retry.js"; import { requestRestartInterruptedTurnRetry } from "./heartbeat/restart-interrupted-turn.js"; +import { startTaskDrain, stopTaskDrain } from "./task-admission.js"; import { activityLog } from "@paperclipai/db"; import { readFreshWorkspaceReadiness } from "./fresh-workspace-continuation.js"; import { materializeIsolatedTaskDirectory } from "./isolated-task-directory.js"; @@ -378,6 +379,146 @@ describe("automatic retained workspace recovery", () => { expect(successors[0].contextSnapshot).toMatchObject({ previousRunId: f.runId, taskRetryRequestId: receipt.requestId }); }, 30_000); + // S54: the automatic Retry's own run then lost its runner too. Saved-work + // recovery recovered its files and the hold offered Retry, but a click was + // admitted and never started a run; a second click joined the stuck intent. + async function failRetrySuccessorWithLostRunner(f: Awaited>) { + const [successor] = await queuedSuccessors(f.companyId); + const runId = successor.id; + await db.update(heartbeatRuns).set({ status: "failed", runtimeMode: "native", nativeIssueId: f.issueId, issueId: f.issueId, + errorCode: "setup_failed", error: "native_remote_recovery_lease_mismatch", startedAt: new Date(Date.now() - 30_000), + finishedAt: new Date(), responsibleUserId: "local-board" }).where(eq(heartbeatRuns.id, runId)); + await db.update(agentWakeupRequests).set({ status: "failed", runId, finishedAt: new Date() }) + .where(eq(agentWakeupRequests.id, successor.wakeupRequestId!)); + await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: f.issueId, runId, phase: "terminal_failure" }); + for (const eventType of ["turn.started", "turn.completed"]) + await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId, agentId: f.agentId, eventType, + stream: "system", level: "info", message: eventType }); + const [environment] = await db.insert(environments).values({ name: runId, driver: "sandbox" }).returning(); + const providerLeaseId = randomUUID(), remote = path.join(root, runId, "remote"); + const [lease] = await db.insert(environmentLeases).values({ companyId: f.companyId, issueId: f.issueId, heartbeatRunId: runId, + environmentId: environment.id, provider: "daytona", providerLeaseId, leasePolicy: "retain_on_failure", status: "released", + cleanupStatus: "success", releasedAt: new Date(), + metadata: { remoteCwd: remote, driver: "sandbox", pluginId: randomUUID(), sandboxProviderPlugin: true } }).returning(); + const target: AdapterSandboxExecutionTarget = { ...targets.get(f.lease.id)!, leaseId: lease.id, remoteCwd: remote, + sandboxLeaseAcquisition: { outcome: "created", providerLeaseId } }; + targets.set(lease.id, target); + expect(await prepareNativeWorkspaceSync({ db, companyId: f.companyId, workspaceId: f.issueId, workspaceLocalDir: f.host, + runId, lease: lease as EnvironmentLease, target, resourceDisposition: "destroy" })).not.toBeNull(); + await writeFile(path.join(remote, "retry-work.txt"), "retry turn work\n"); + const recovery = { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [lease.id] }; + await db.update(heartbeatRuns).set({ resultJson: { workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: recovery, + conversationContinuation: "continue_conversation_v1" } }).where(eq(heartbeatRuns.id, runId)); + await db.update(environmentLeases).set({ metadata: { ...lease.metadata, + remoteExecutionTermination: remoteTerminationReceipt(lease, { providerLeaseId, state: "stopped" }) } }).where(eq(environmentLeases.id, lease.id)); + const [hold] = await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: f.issueId, kind: "active_run_watchdog", + ownerType: "board", cause: "native_continuation_requires_reconciliation", fingerprint: runId, nextAction: "Manual repair required", + evidence: { runId, workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: recovery } }).returning(); + const environmentRuntime = environmentRuntimeService(db, { drivers: [{ driver: "sandbox", + acquireRunLease: vi.fn(), releaseRunLease: vi.fn(), + resumeRunLease: vi.fn(async (input: { lease: EnvironmentLease }) => ({ providerLeaseId: input.lease.providerLeaseId, + metadata: { remoteCwd: (input.lease.metadata as { remoteCwd?: string }).remoteCwd } })), + retryPendingSandboxTeardown: vi.fn(async (input: { lease: EnvironmentLease }) => ({ providerLeaseId: input.lease.providerLeaseId, state: "stopped" })), + resolveCapabilities: async () => classifyEnvironmentCapabilities({ verifiedMethods: [], declared: null, narrowing: null }), + }] }); + return { runId, hold, environmentRuntime, claim: { ...f.claim, runId, remote: { providerLeaseId, remoteCwd: remote } } }; + } + + async function recoverRetryLostRunner() { + const f = await restartAllocationLostSource("settled", "runner_remote_process_identity_unavailable"); + const service = heartbeatService(db, { environmentRuntime: f.environmentRuntime }); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }); + expect(await requestRestartInterruptedTurnRetry(db, f.claim)).toMatchObject({ requested: true }); + await vi.waitFor(async () => { + await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, current => service.resumeRemoteStopComments(current)) ?? []); + expect(readSystemRecovery((await action(f.actionId)).evidence)?.state).toBe("complete"); + }, { timeout: 10_000, interval: 10 }); + await vi.waitFor(async () => expect(await queuedSuccessors(f.companyId)).toHaveLength(1), { timeout: 10_000, interval: 50 }); + const second = await failRetrySuccessorWithLostRunner(f); + const secondService = heartbeatService(db, { environmentRuntime: second.environmentRuntime }); + // That automatic Retry's own loss is left to the user's visible Retry. + expect(await requestRestartInterruptedTurnRetry(db, second.claim)).toEqual({ requested: false, reason: "automatic_retry_used" }); + await vi.waitFor(async () => { + await Promise.all(await reconcileRetainedWorkspaces(db, second.environmentRuntime, current => secondService.resumeRemoteStopComments(current)) ?? []); + expect(readSystemRecovery((await action(second.hold.id)).evidence)?.state).toBe("complete"); + }, { timeout: 10_000, interval: 10 }); + expect(await queuedSuccessors(f.companyId)).toHaveLength(0); + const issue = (await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]; + const blocker = await getExecutionBlocker(db, f.companyId, f.issueId); + expect(blocker).toMatchObject({ runId: second.runId, cause: "native_continuation_requires_reconciliation", canRetry: true, + nextAction: "Saved work recovered. Retry or send a message to continue." }); + expect(await taskInterruptionPresentation(db, issue)).toMatchObject({ canRetry: true, observedRunId: second.runId }); + return { ...f, ...second, service: secondService, issue }; + } + + it("starts exactly one run when Retry is pressed after the automatic Retry's own run lost its runner", async () => { + const f = await recoverRetryLostRunner(); + const receipt = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issueId, actorId: "local-board", + actorSource: "local_implicit", idempotencyKey: randomUUID(), observedRunId: f.runId }); + expect(receipt).toMatchObject({ accepted: true }); + await f.service.resumeTaskRetry(f.companyId, receipt.requestId!); + const successors = await queuedSuccessors(f.companyId); + expect(successors).toHaveLength(1); + expect(successors[0].contextSnapshot).toMatchObject({ previousRunId: f.runId, taskRetryRequestId: receipt.requestId }); + // A replayed dispatch and a second click do not start another run. + await f.service.resumeTaskRetry(f.companyId, receipt.requestId!); + await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issueId, actorId: "local-board", + actorSource: "local_implicit", idempotencyKey: randomUUID(), observedRunId: f.runId }); + expect(await queuedSuccessors(f.companyId)).toHaveLength(1); + }, 60_000); + + it("does not let a Retry that could not start swallow a later Retry", async () => { + const f = await recoverRetryLostRunner(); + // The first click's intent is declined at dispatch: its requester no longer + // has access. It must name that gate instead of waiting silently. + const first = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issueId, actorId: "departed-user", + actorSource: "session", idempotencyKey: randomUUID(), observedRunId: f.runId }); + expect(first).toMatchObject({ accepted: true }); + await f.service.resumeTaskRetry(f.companyId, first.requestId!); + expect(await queuedSuccessors(f.companyId)).toHaveLength(0); + const [declined] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, first.requestId!)); + expect(declined).toMatchObject({ status: "deferred_issue_execution", runId: null, + payload: { executionWait: { reason: "retry_authority_denied" } } }); + expect(await taskInterruptionPresentation(db, f.issue)).toMatchObject({ state: "failed", canRetry: true, + retryWait: { reason: "retry_authority_denied" } }); + // A later click is a fresh attempt under the clicker's authority, not the stuck intent. + const second = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issueId, actorId: "local-board", + actorSource: "local_implicit", idempotencyKey: randomUUID(), observedRunId: f.runId }); + expect(second).toMatchObject({ accepted: true }); + expect(second.requestId).not.toBe(first.requestId); + await f.service.resumeTaskRetry(f.companyId, second.requestId!); + // A late dispatch of the replaced intent cannot start a second run. + await f.service.resumeTaskRetry(f.companyId, first.requestId!); + const successors = await queuedSuccessors(f.companyId); + expect(successors).toHaveLength(1); + expect(successors[0].contextSnapshot).toMatchObject({ previousRunId: f.runId, taskRetryRequestId: second.requestId }); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, first.requestId!)))[0]) + .toMatchObject({ status: "skipped", error: "task_retry_replaced", runId: null }); + const activity = await db.select().from(activityLog).where(and(eq(activityLog.companyId, f.companyId), + eq(activityLog.action, "issue.retry_requested"), eq(activityLog.actorId, "local-board"))); + expect(activity.at(-1)?.details).toMatchObject({ requestId: second.requestId, replacedRequestId: first.requestId }); + }, 60_000); + + it("dispatches a Retry accepted during the pre-deploy drain once the restarted server resumes", async () => { + const f = await recoverRetryLostRunner(); + startTaskDrain({ ttlMs: null }); + let receipt: Awaited>; + try { + receipt = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issueId, actorId: "local-board", + actorSource: "local_implicit", idempotencyKey: randomUUID(), observedRunId: f.runId }); + expect(receipt).toMatchObject({ accepted: true }); + await f.service.resumeTaskRetry(f.companyId, receipt.requestId!); + expect(await queuedSuccessors(f.companyId)).toHaveLength(0); + } finally { stopTaskDrain(); } + // The drain is process-local; a restarted server redelivers the saved intent. + const restarted = heartbeatService(db, { environmentRuntime: f.environmentRuntime }); + await restarted.resumeExecutionWaitComments({ companyId: f.companyId }); + await restarted.resumeExecutionWaitComments({ companyId: f.companyId }); + const successors = await queuedSuccessors(f.companyId); + expect(successors).toHaveLength(1); + expect(successors[0].contextSnapshot).toMatchObject({ previousRunId: f.runId, taskRetryRequestId: receipt!.requestId }); + }, 60_000); + async function failedWarmSource(exportOnly = false, hops = 0) { const f = hops ? await warmHandoffSource(hops) : await seed("native", true, "keep_running"); if (hops) await db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, f.runId));