diff --git a/server/src/services/execution-continuation.ts b/server/src/services/execution-continuation.ts index c4cd1e95819..332d3ec335f 100644 --- a/server/src/services/execution-continuation.ts +++ b/server/src/services/execution-continuation.ts @@ -458,7 +458,6 @@ export async function buildExecutionContinuation(input: { eq(agentWakeupRequests.agentId, input.agentId), eq(agentWakeupRequests.reason, "task_retry"), eq(agentWakeupRequests.requestedByActorType, "user"), eq(agentWakeupRequests.status, "coalesced"), sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`, - input.runId ? eq(agentWakeupRequests.runId, input.runId) : undefined, )) : []; const taskRetry = readTaskRetryIntent(taskRetryWake?.payload); const retryAuthorization = taskRetry && taskRetry.generationId === taskRetryWake?.id @@ -485,9 +484,15 @@ export async function buildExecutionContinuation(input: { input.runId ? eq(agentWakeupRequests.runId, input.runId) : undefined, )) : []; const authorization = continuationAuthorizations.find(value => taskRetryId - ? value.taskRetryRequestId === taskRetryId && taskRetryWake?.runId === value.runId && - taskRetryWake.requestedByActorId === value.actorId && - object(taskRetryWake.payload?.taskRetry).sourceRunId === value.previousRunId + ? value.taskRetryRequestId === taskRetryId && taskRetryWake?.requestedByActorId === value.actorId && ( + // The Retry receipt stays bound to the run it admitted. A bounded + // automatic retry of that run carries its own admission receipt. + value.automaticRetry + ? taskRetry?.generationId === taskRetryWake.id && taskRetry.actorId === value.actorId && + object(value.automaticRetry).sourceRunId === explicitUserSource && issue.executionRunId === input.runId && + priorRuns.some(run => run.id === value.runId && run.retryOfRunId === explicitUserSource) + : taskRetryWake.runId === value.runId && + object(taskRetryWake.payload?.taskRetry).sourceRunId === value.previousRunId) : failedRunId ? value.failedRunId === failedRunId && retryWakes.some(wake => wake.runId === value.runId && wake.requestedByActorId === value.actorId && diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index 13075ff1f59..ddf3ae1c46a 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -576,15 +576,81 @@ export async function admitExplicitNativeContinuation(input: { return continuation; } +type ContinuationRetryAuthority = { + continuation: { commentId: string } | { commentId: null; taskRetryRequestId: string }; + actorId: string; sourceAuthorizationId: string; evidence: Record; +}; + +async function readCommentRetryAuthority(db: Db, input: { + companyId: string; issueId: string; parentRunId: string; explicit: Record; +}): Promise { + const { companyId, issueId, explicit } = input; + if (!z.string().guid().safeParse(explicit.commentId).success) return null; + const commentId = explicit.commentId as string; + const [comment] = await db.select().from(issueComments).where(and( + eq(issueComments.companyId, companyId), eq(issueComments.issueId, issueId), + eq(issueComments.id, commentId), eq(issueComments.authorType, "user"), + isNull(issueComments.createdByRunId), isNull(issueComments.deletedAt), + )); + if (!comment?.authorUserId || !comment.body.trim()) return null; + const commentBodyHash = createHash("sha256").update(comment.body).digest("hex"); + const receipts = await db.select().from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.sourceIssueId, issueId), + eq(issueRecoveryActions.status, "resolved"), + sql`${issueRecoveryActions.evidence}->'explicitUserContinuation'->>'runId' = ${input.parentRunId}`, + )); + const receipt = receipts.find(row => { + const auth = row.evidence.explicitUserContinuation as Record | undefined; + return auth && auth.previousRunId === explicit.previousRunId && auth.commentId === commentId && + auth.actorId === comment.authorUserId && !auth.failedRunId && !auth.queuedCommentInterruptId && + auth.commentUpdatedAt === comment.updatedAt.toISOString() && auth.commentBodyHash === commentBodyHash; + }); + return receipt ? { continuation: { commentId }, actorId: comment.authorUserId, sourceAuthorizationId: receipt.id, + evidence: { commentUpdatedAt: comment.updatedAt.toISOString(), commentBodyHash } } : null; +} + +/** A task Retry admits one plain continuation in the saved workspace. When that + * run fails transiently, its bounded retry keeps the same actor and receipt; + * chat, queued-message and fresh-workspace Retries keep their own contracts. */ +async function readTaskRetryRetryAuthority(db: Db, input: { + companyId: string; issueId: string; agentId: string; parentRunId: string; explicit: Record; +}): Promise { + const { companyId, issueId, explicit } = input; + const requestId = z.string().guid().safeParse(explicit.taskRetryRequestId); + if (explicit.commentId !== null || !requestId.success || explicit.queuedCommentInterruptId || explicit.readiness) return null; + const [wake] = await db.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.id, requestId.data), + eq(agentWakeupRequests.agentId, input.agentId), eq(agentWakeupRequests.reason, "task_retry"), + eq(agentWakeupRequests.requestedByActorType, "user"), eq(agentWakeupRequests.status, "coalesced"), + sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`, + )); + const intent = readTaskRetryIntent(wake?.payload); + if (!wake?.requestedByActorId || !wake.runId || !intent || intent.generationId !== wake.id || + intent.actorId !== wake.requestedByActorId || intent.strategy !== "continue" || + intent.chatActionId || intent.queuedChatRequestId || intent.canonicalIntentId) return null; + // The admitted run's authority may be its action ledger, an earlier + // automatic-retry receipt, or the Retry receipt itself. Dispatch validation + // of the parent below proves which one applies. + const [receipt] = await db.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.sourceIssueId, issueId), + eq(issueRecoveryActions.status, "resolved"), + sql`${issueRecoveryActions.evidence}->'explicitUserContinuation'->>'runId' = ${input.parentRunId}`, + sql`${issueRecoveryActions.evidence}->'explicitUserContinuation'->>'taskRetryRequestId' = ${wake.id}`, + )).limit(1); + return { continuation: { commentId: null, taskRetryRequestId: wake.id }, actorId: wake.requestedByActorId, + sourceAuthorizationId: receipt?.id ?? wake.id, evidence: {} }; +} + /** Re-admit a bounded automatic retry without lending it its parent's receipt. * The caller holds the task lock and creates the successor in this transaction. - * Only unchanged user messages are renewable; Retry/Interrupt intents keep their - * separate one-run contracts. The scheduler still owns retry and cleanup policy. + * Unchanged user messages and plain task Retries are renewable; Interrupt and + * failed-run retry intents keep their separate one-run contracts. The + * scheduler still owns retry and cleanup policy. */ export async function admitExplicitContinuationRetry(input: { db: Db; companyId: string; issueId: string; agentId: string; parentRunId: string; successorRunId: string; now: Date; -}): Promise<{ previousRunId: string; commentId: string } | null> { +}): Promise<{ previousRunId: string; commentId: string | null; taskRetryRequestId?: string } | null> { const { db, companyId, issueId, agentId } = input; const [task] = await db.select().from(issues).where(and( eq(issues.companyId, companyId), eq(issues.id, issueId), @@ -602,28 +668,11 @@ export async function admitExplicitContinuationRetry(input: { await getExecutionBlocker(db, companyId, issueId)) return null; const context = parent.contextSnapshot; const explicit = context.explicitUserContinuation as Record | undefined; - if (!explicit || !z.string().guid().safeParse(explicit.previousRunId).success || - !z.string().guid().safeParse(explicit.commentId).success || explicit.failedRunId) return null; - const commentId = explicit.commentId as string; - const [comment] = await db.select().from(issueComments).where(and( - eq(issueComments.companyId, companyId), eq(issueComments.issueId, issueId), - eq(issueComments.id, commentId), eq(issueComments.authorType, "user"), - isNull(issueComments.createdByRunId), isNull(issueComments.deletedAt), - )); - if (!comment?.authorUserId || !comment.body.trim()) return null; - const receipts = await db.select().from(issueRecoveryActions).where(and( - eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.sourceIssueId, issueId), - eq(issueRecoveryActions.status, "resolved"), - sql`${issueRecoveryActions.evidence}->'explicitUserContinuation'->>'runId' = ${parent.id}`, - )); - const receipt = receipts.find(row => { - const auth = row.evidence.explicitUserContinuation as Record | undefined; - return auth && auth.previousRunId === explicit.previousRunId && auth.commentId === commentId && - auth.actorId === comment.authorUserId && !auth.failedRunId && !auth.queuedCommentInterruptId && - auth.commentUpdatedAt === comment.updatedAt.toISOString() && - auth.commentBodyHash === createHash("sha256").update(comment.body).digest("hex"); - }); - if (!receipt) return null; + if (!explicit || !z.string().guid().safeParse(explicit.previousRunId).success || explicit.failedRunId) return null; + const authority = explicit.commentId === null + ? await readTaskRetryRetryAuthority(db, { companyId, issueId, agentId, parentRunId: parent.id, explicit }) + : await readCommentRetryAuthority(db, { companyId, issueId, parentRunId: parent.id, explicit }); + if (!authority) return null; const [superseding] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( eq(heartbeatRuns.companyId, companyId), or(eq(heartbeatRuns.nativeIssueId, issueId), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`), @@ -646,18 +695,18 @@ export async function admitExplicitContinuationRetry(input: { "continuation_source_context_missing", "continuation_task_ownership_changed"].includes(error.message)) return null; throw error; } - const continuation = { previousRunId: parent.id, commentId }; + const continuation = { previousRunId: parent.id, ...authority.continuation }; await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issueId, kind: "active_run_watchdog", cause: "explicit_user_continuation_retry", fingerprint: input.successorRunId, status: "resolved", outcome: "retry_authorized", resolvedAt: input.now, - nextAction: "The bounded retry has its own authorization for the unchanged user message.", + nextAction: authority.continuation.commentId + ? "The bounded retry has its own authorization for the unchanged user message." + : "The bounded retry has its own authorization for the same task Retry.", evidence: { explicitUserContinuation: { - ...continuation, actorId: comment.authorUserId, runId: input.successorRunId, - commentUpdatedAt: comment.updatedAt.toISOString(), - commentBodyHash: createHash("sha256").update(comment.body).digest("hex"), + ...continuation, actorId: authority.actorId, runId: input.successorRunId, ...authority.evidence, recordedAt: input.now.toISOString(), automaticRetry: { - sourceRunId: parent.id, sourceAuthorizationId: receipt.id, + sourceRunId: parent.id, sourceAuthorizationId: authority.sourceAuthorizationId, }, } }, }); diff --git a/server/src/services/task-retry.test.ts b/server/src/services/task-retry.test.ts index d7e02d1d339..c1a29ddc465 100644 --- a/server/src/services/task-retry.test.ts +++ b/server/src/services/task-retry.test.ts @@ -83,6 +83,112 @@ describe("durable task Retry admission", () => { expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, source.id)))[0]) .toMatchObject({ status: "failed", resultJson: source.resultJson }); }); + // A run admitted by Retry can fail before provider dispatch, for example when + // a saved-workspace download is reset. Its bounded transient retry carries + // the same Retry authority, recorded for the new run. + async function seedFailedRetryAdmittedRun() { + const input = await seed(); + await db.update(issues).set({ status: "todo" }).where(eq(issues.id, input.issueId)); + const [source] = await db.update(heartbeatRuns).set({ runtimeMode: "legacy", errorCode: "setup_failed", + executionStage: "preparing", runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } }, + resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }, + }).where(eq(heartbeatRuns.id, input.observedRunId)).returning(); + const receipt = await requestTaskRetry(input), parentId = randomUUID(); + const continuation = await db.transaction(async tx => { + await tx.select().from(issues).where(eq(issues.id, input.issueId)).for("update"); + const result = await admitExplicitNativeContinuation({ db: tx as unknown as Db, + companyId: input.companyId, issueId: input.issueId, agentId: source.agentId, + actorId: input.actorId, actorType: "user", reason: "task_retry", commentId: null, + successorRunId: parentId, taskRetryRequestId: receipt.requestId!, + }); + await tx.insert(heartbeatRuns).values({ id: parentId, companyId: input.companyId, agentId: source.agentId, + issueId: input.issueId, status: "queued", contextSnapshot: { issueId: input.issueId } }); + await tx.update(agentWakeupRequests).set({ status: "coalesced", runId: parentId }) + .where(eq(agentWakeupRequests.id, receipt.requestId!)); + return result; + }); + expect(continuation).toMatchObject({ previousRunId: source.id, commentId: null, taskRetryRequestId: receipt.requestId }); + const now = new Date(); + const [parent] = await db.update(heartbeatRuns).set({ runtimeMode: "legacy", status: "failed", responsibleUserId: input.actorId, + errorCode: "remote_workspace_restore_failed", executionStage: "preparing", startedAt: now, finishedAt: now, + runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } }, + resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false }, + remoteWorkspaceRestore: { step: "snapshot_download", retryable: true } }, + contextSnapshot: { issueId: input.issueId, wakeReason: "task_retry", taskRetryRequestId: receipt.requestId, + previousRunId: source.id, explicitUserContinuation: continuation }, + }).where(eq(heartbeatRuns.id, parentId)).returning(); + await db.update(issues).set({ status: "in_progress", executionRunId: parent.id }).where(eq(issues.id, input.issueId)); + return { input, source, parent, now, retryRequestId: receipt.requestId! }; + } + const dispatchRun = (input: { companyId: string; issueId: string }, run: typeof heartbeatRuns.$inferSelect) => + buildExecutionContinuation({ db, companyId: input.companyId, issueId: input.issueId, agentId: run.agentId, + runId: run.id, context: run.contextSnapshot!, summary: null, exposeLowTrustRaw: false }); + it("retries a transient preparation failure of a Retry-admitted run under the same Retry authority", async () => { + const f = await seedFailedRetryAdmittedRun(); + await expect(dispatchRun(f.input, f.parent)).resolves.toMatchObject({ interruptedRunId: f.source.id }); + const heartbeat = heartbeatService(db); + const first = await heartbeat.scheduleBoundedRetry(f.parent.id, { now: f.now, delayMs: 1000 }); + expect(first.outcome).toBe("scheduled"); + if (first.outcome !== "scheduled") return; + expect(first.run).toMatchObject({ retryOfRunId: f.parent.id, scheduledRetryAttempt: 1 }); + expect(first.run.contextSnapshot?.explicitUserContinuation).toEqual({ + previousRunId: f.parent.id, commentId: null, taskRetryRequestId: f.retryRequestId }); + // The Retry receipt stays bound to the run it admitted. + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.retryRequestId)))[0]) + .toMatchObject({ status: "coalesced", runId: f.parent.id }); + await expect(dispatchRun(f.input, first.run)).resolves.toMatchObject({ interruptedRunId: f.parent.id }); + // A repeated request adopts the same successor without another receipt. + expect(await heartbeat.scheduleBoundedRetry(f.parent.id, { now: f.now })).toMatchObject({ + outcome: "scheduled", run: { id: first.run.id } }); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.input.issueId))) + .toHaveLength(1); + // A second transient failure retries from its immediate parent, then the budget ends. + const later = new Date(f.now.getTime() + 2000); + const [failedFirst] = await db.update(heartbeatRuns).set({ status: "failed", errorCode: "remote_workspace_restore_failed", + startedAt: later, finishedAt: later, resultJson: f.parent.resultJson }).where(eq(heartbeatRuns.id, first.run.id)).returning(); + await db.update(issues).set({ executionRunId: failedFirst.id }).where(eq(issues.id, f.input.issueId)); + const second = await heartbeat.scheduleBoundedRetry(failedFirst.id, { now: later }); + expect(second.outcome).toBe("scheduled"); + if (second.outcome !== "scheduled") return; + expect(second.run).toMatchObject({ retryOfRunId: failedFirst.id, scheduledRetryAttempt: 2 }); + await expect(dispatchRun(f.input, second.run)).resolves.toMatchObject({ interruptedRunId: failedFirst.id }); + await db.update(heartbeatRuns).set({ status: "failed", errorCode: "remote_workspace_restore_failed", + startedAt: later, finishedAt: later }).where(eq(heartbeatRuns.id, second.run.id)); + await db.update(issues).set({ executionRunId: second.run.id }).where(eq(issues.id, f.input.issueId)); + expect(await heartbeat.scheduleBoundedRetry(second.run.id, { now: later })).toMatchObject({ outcome: "retry_exhausted" }); + // Removing the successor's own receipt reproduces the dispatch refusal. + const receipts = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.input.issueId)); + expect(receipts).toHaveLength(2); + await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.id, + receipts.find(row => (row.evidence.explicitUserContinuation as Record).runId === second.run.id)!.id)); + await expect(dispatchRun(f.input, second.run)).rejects.toThrow("continuation_user_authorization_missing"); + }); + it.each(["wrong_actor", "unadmitted", "wrong_generation", "other_run", "chat_retry", "fresh_workspace", "reassigned", "superseded"] as const)( + "does not lend Retry authority to an automatic retry after %s", async kind => { + const f = await seedFailedRetryAdmittedRun(); + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.retryRequestId)); + const taskRetry = wake.payload!.taskRetry as Record; + if (["wrong_actor", "unadmitted", "wrong_generation", "other_run", "chat_retry"].includes(kind)) + await db.update(agentWakeupRequests).set({ + ...(kind === "wrong_actor" ? { requestedByActorId: "other-user" } : {}), + ...(kind === "unadmitted" ? { status: "deferred_issue_execution" } : {}), + ...(kind === "other_run" ? { runId: f.source.id } : {}), + ...(kind === "wrong_generation" ? { payload: { ...wake.payload, taskRetry: { ...taskRetry, generationId: randomUUID() } } } : {}), + ...(kind === "chat_retry" ? { payload: { ...wake.payload, taskRetry: { ...taskRetry, chatActionId: randomUUID() } } } : {}), + }).where(eq(agentWakeupRequests.id, f.retryRequestId)); + if (kind === "fresh_workspace") await db.update(heartbeatRuns).set({ contextSnapshot: { ...f.parent.contextSnapshot, + explicitUserContinuation: { ...(f.parent.contextSnapshot!.explicitUserContinuation as Record), + readiness: { mode: "fresh_isolated_workspace", retryIntentId: f.retryRequestId, sourceRunId: f.source.id, retainedSourceIds: [] } } }, + }).where(eq(heartbeatRuns.id, f.parent.id)); + if (kind === "reassigned") await db.update(issues).set({ assigneeAgentId: null }).where(eq(issues.id, f.input.issueId)); + if (kind === "superseded") await db.insert(heartbeatRuns).values({ companyId: f.input.companyId, agentId: f.parent.agentId, + status: "succeeded", contextSnapshot: { issueId: f.input.issueId }, createdAt: new Date(f.now.getTime() + 1000) }); + const result = await heartbeatService(db).scheduleBoundedRetry(f.parent.id, { now: f.now }); + expect(result.outcome).not.toBe("scheduled"); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, f.parent.id))).toHaveLength(0); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.input.issueId))) + .toHaveLength(0); + }); it("serializes concurrent clicks and remembers each key after completion", async () => { const input = await seed(); const second = { ...input, idempotencyKey: randomUUID() };