diff --git a/apps/web/src/app/(sandbox)/task/[taskId]/TaskInputStack.tsx b/apps/web/src/app/(sandbox)/task/[taskId]/TaskInputStack.tsx index f9275bfe5..b8782886c 100644 --- a/apps/web/src/app/(sandbox)/task/[taskId]/TaskInputStack.tsx +++ b/apps/web/src/app/(sandbox)/task/[taskId]/TaskInputStack.tsx @@ -57,7 +57,16 @@ export function TaskInputStack({
diff --git a/apps/web/src/app/(sandbox)/task/[taskId]/page.tsx b/apps/web/src/app/(sandbox)/task/[taskId]/page.tsx index fe4441366..9125a9585 100644 --- a/apps/web/src/app/(sandbox)/task/[taskId]/page.tsx +++ b/apps/web/src/app/(sandbox)/task/[taskId]/page.tsx @@ -69,6 +69,12 @@ export default function SandboxPage() { const hasArtifacts = session.artifacts.length > 0; const hasVisibleSessionPrompt = session.prompt?.visibleInTranscript !== false && session.prompt != null; + const startupPrompt = hasVisibleSessionPrompt + ? { + text: session.prompt?.text, + images: session.prompt?.images, + } + : null; const shouldRenderBootingTranscript = sessionState === 'booting' && (hasTranscriptHistory || hasVisibleSessionPrompt); @@ -262,7 +268,13 @@ export default function SandboxPage() { : null + taskRun ? ( + + ) : null } /> @@ -275,7 +287,9 @@ export default function SandboxPage() {
{session.draftPrompt && ( @@ -291,7 +305,9 @@ export default function SandboxPage() {
{session.draftPrompt && ( diff --git a/apps/web/src/app/(sandbox)/task/[taskId]/startup/Startup.tsx b/apps/web/src/app/(sandbox)/task/[taskId]/startup/Startup.tsx index 31643af0e..fdf0e278b 100644 --- a/apps/web/src/app/(sandbox)/task/[taskId]/startup/Startup.tsx +++ b/apps/web/src/app/(sandbox)/task/[taskId]/startup/Startup.tsx @@ -3,24 +3,112 @@ import { useCallback } from 'react'; import { SSEProvider } from 'react-hooks-sse'; -import { TaskPayloadKind, type RunStatus } from '@roomote/types'; +import { + RunStatus, + TaskPayloadKind, + type RunStatus as RunStatusValue, +} from '@roomote/types'; import type { TaskRun } from '@roomote/db'; import { useRestoreTaskRunSnapshot } from '@/hooks/snapshots'; +import { useRetryFailedTaskStart } from '@/hooks/task-runs'; import { getTaskRunError } from '@/lib/task-run-errors'; -import { StartupFailureMessage, StartupSequence } from './StartupMessage'; +import { + StartupFailureMessage, + StartupSequence, + type StartupPromptPreview, + type StartupRetryAction, +} from './StartupMessage'; import { useStartupProgress } from './useStartupProgress'; +/** + * Client-side eligibility for start retry. Keep in lockstep with + * `isRelaunchableFailedStartPayloadKind` in packages/cloud-agents. + */ +function canRelaunchFailedStart( + taskRun: Pick, +): boolean { + if (taskRun.status !== RunStatus.Failed) { + return false; + } + + switch (taskRun.payloadKind) { + case TaskPayloadKind.StandardTask: + case TaskPayloadKind.Scan: + case TaskPayloadKind.SlackAppMention: + case TaskPayloadKind.LinearAgentSession: + case TaskPayloadKind.GithubPrReviewFollowUp: + case TaskPayloadKind.McpRecommendations: + return true; + default: + return false; + } +} + +function buildRetryAction(params: { + taskId: string; + taskRun?: + | Pick< + TaskRun, + 'id' | 'payloadKind' | 'status' | 'sourceRunId' | 'sourceSnapshotId' + > + | TaskRun; + restoreSnapshot: ReturnType; + retryFailedStart: ReturnType; +}): StartupRetryAction | undefined { + const { taskId, taskRun, restoreSnapshot, retryFailedStart } = params; + + if (!taskRun) { + return undefined; + } + + const canRetryResume = + taskRun.payloadKind === TaskPayloadKind.SnapshotResume && + typeof taskRun.sourceSnapshotId === 'string' && + taskRun.sourceSnapshotId.length > 0 && + typeof taskRun.sourceRunId === 'number'; + + if (canRetryResume) { + return { + label: 'Retry resume', + onClick: () => + restoreSnapshot.mutate({ + sourceSnapshotId: taskRun.sourceSnapshotId!, + sourceRunId: taskRun.sourceRunId!, + }), + pending: restoreSnapshot.isPending, + }; + } + + if (canRelaunchFailedStart(taskRun)) { + return { + label: 'Retry', + onClick: () => + retryFailedStart.mutate({ + taskId, + runId: taskRun.id, + }), + pending: retryFailedStart.isPending, + }; + } + + return undefined; +} + interface StartupProps { runId: number; + taskId: string; initialTaskRun?: TaskRun; - onStatusChange?: (status: RunStatus) => void; + prompt?: StartupPromptPreview | null; + onStatusChange?: (status: RunStatusValue) => void; } export const Startup = ({ runId, + taskId, initialTaskRun, + prompt, onStatusChange, }: StartupProps) => { const eventSource = useCallback(() => { @@ -37,7 +125,9 @@ export const Startup = ({ @@ -46,26 +136,25 @@ export const Startup = ({ interface StartupInnerProps { runId: number; + taskId: string; initialTaskRun?: TaskRun; - onStatusChange?: (status: RunStatus) => void; + prompt?: StartupPromptPreview | null; + onStatusChange?: (status: RunStatusValue) => void; } const StartupInner = ({ runId, + taskId, initialTaskRun, + prompt, onStatusChange, }: StartupInnerProps) => { const restoreSnapshot = useRestoreTaskRunSnapshot(); + const retryFailedStart = useRetryFailedTaskStart(); const { steps, error, showLogs, sandboxLogs, logsConnected, logsError } = useStartupProgress({ runId, initialTaskRun, onStatusChange }); - const canRetryResume = - initialTaskRun?.payloadKind === TaskPayloadKind.SnapshotResume && - typeof initialTaskRun.sourceSnapshotId === 'string' && - initialTaskRun.sourceSnapshotId.length > 0 && - typeof initialTaskRun.sourceRunId === 'number'; - return ( - restoreSnapshot.mutate({ - sourceSnapshotId: initialTaskRun.sourceSnapshotId!, - sourceRunId: initialTaskRun.sourceRunId!, - }), - pending: restoreSnapshot.isPending, - } - : undefined - } + prompt={prompt} + retryAction={buildRetryAction({ + taskId, + taskRun: initialTaskRun, + restoreSnapshot, + retryFailedStart, + })} /> ); }; interface SnapshotResumeFailureFooterProps { + taskId: string; taskRun: Pick< TaskRun, + | 'id' | 'error' | 'result' | 'sourceRunId' @@ -99,35 +185,28 @@ interface SnapshotResumeFailureFooterProps { | 'status' | 'payloadKind' >; + prompt?: StartupPromptPreview | null; } export const SnapshotResumeFailureFooter = ({ + taskId, taskRun, + prompt, }: SnapshotResumeFailureFooterProps) => { const restoreSnapshot = useRestoreTaskRunSnapshot(); - - const canRetryResume = - taskRun.payloadKind === TaskPayloadKind.SnapshotResume && - typeof taskRun.sourceSnapshotId === 'string' && - taskRun.sourceSnapshotId.length > 0 && - typeof taskRun.sourceRunId === 'number'; + const retryFailedStart = useRetryFailedTaskStart(); return ( - restoreSnapshot.mutate({ - sourceSnapshotId: taskRun.sourceSnapshotId!, - sourceRunId: taskRun.sourceRunId!, - }), - pending: restoreSnapshot.isPending, - } - : undefined - } + prompt={prompt} + retryAction={buildRetryAction({ + taskId, + taskRun, + restoreSnapshot, + retryFailedStart, + })} /> ); }; diff --git a/apps/web/src/app/(sandbox)/task/[taskId]/startup/StartupMessage.client.test.tsx b/apps/web/src/app/(sandbox)/task/[taskId]/startup/StartupMessage.client.test.tsx index 0aa0375f2..a693ec65d 100644 --- a/apps/web/src/app/(sandbox)/task/[taskId]/startup/StartupMessage.client.test.tsx +++ b/apps/web/src/app/(sandbox)/task/[taskId]/startup/StartupMessage.client.test.tsx @@ -128,14 +128,14 @@ describe('StartupSequence', () => { expect(screen.getByText('Warming up my GPUs')).toBeInTheDocument(); }); - it('shows a retry resume button for startup failures when a retry action is provided', () => { + it('shows a retry button for startup failures when a retry action is provided', () => { const onClick = vi.fn(); render( , ); @@ -146,6 +146,42 @@ describe('StartupSequence', () => { expect(onClick).toHaveBeenCalledTimes(1); }); + it('shows the stored prompt and default Retry label for failed environment starts', () => { + const onClick = vi.fn(); + + render( + , + ); + + expect(screen.getByText('Your prompt')).toBeInTheDocument(); + expect( + screen.getByText('Add multi-camera clip switching'), + ).toBeInTheDocument(); + + const button = screen.getByRole('button', { name: 'Retry' }); + fireEvent.click(button); + expect(onClick).toHaveBeenCalledTimes(1); + }); + + it('shows image attachments for failed starts without a displayable error string', () => { + render( + , + ); + + expect(screen.getByText('Your prompt')).toBeInTheDocument(); + expect( + screen.getByRole('link', { name: 'View attachment' }), + ).toHaveAttribute('href', 'https://example.com/shot.png'); + }); + it('uses a flex-bounded scroll container when startup content exceeds the available height', () => { const { container } = render( { ); }; +export type StartupRetryAction = { + onClick: () => void; + pending?: boolean; + label?: string; +}; + +export type StartupPromptPreview = { + text?: string; + images?: string[]; +}; + interface StartupErrorMessageProps { status: RunStatus; error?: string; - retryAction?: { - onClick: () => void; - pending?: boolean; - }; + prompt?: StartupPromptPreview | null; + retryAction?: StartupRetryAction; } export const StartupFailureMessage = ({ status, error, + prompt, retryAction, }: StartupErrorMessageProps) => { const isFailed = status === RunStatus.Failed; const isCanceled = status === RunStatus.Canceled; const displayError = getTaskRunErrorDisplayMessage(error); + const promptText = prompt?.text?.trim() || undefined; + const promptImages = prompt?.images?.filter(Boolean) ?? []; + const hasPrompt = Boolean(promptText) || promptImages.length > 0; if ((isFailed || (isCanceled && displayError)) && displayError) { return ( @@ -143,13 +156,34 @@ export const StartupFailureMessage = ({
-
+
There was an error starting this environment:
{displayError}
+ {hasPrompt && ( +
+
Your prompt
+ {promptText && ( +
+ {promptText} +
+ )} + {promptImages.length > 0 && ( +
+ {promptImages.map((image) => ( + + ))} +
+ )} +
+ )} {retryAction && ( -
+
)} @@ -174,7 +208,43 @@ export const StartupFailureMessage = ({
-
There was an error starting this environment:
+
+
There was an error starting this environment:
+ {hasPrompt && ( +
+
Your prompt
+ {promptText && ( +
+ {promptText} +
+ )} + {promptImages.length > 0 && ( +
+ {promptImages.map((image) => ( + + ))} +
+ )} +
+ )} + {retryAction && ( +
+ +
+ )} +
@@ -206,10 +276,8 @@ interface StartupSequenceProps { logs?: SandboxLogEntry[]; logsConnected?: boolean; logsError?: string | null; - retryAction?: { - onClick: () => void; - pending?: boolean; - }; + prompt?: StartupPromptPreview | null; + retryAction?: StartupRetryAction; } export const StartupSequence = ({ @@ -218,6 +286,7 @@ export const StartupSequence = ({ logs, logsConnected = true, logsError = null, + prompt, retryAction, }: StartupSequenceProps) => { const lastStep = steps[steps.length - 1]; @@ -270,6 +339,7 @@ export const StartupSequence = ({
diff --git a/apps/web/src/hooks/task-runs/index.ts b/apps/web/src/hooks/task-runs/index.ts index 9e63a15ac..81d0753cb 100644 --- a/apps/web/src/hooks/task-runs/index.ts +++ b/apps/web/src/hooks/task-runs/index.ts @@ -1,3 +1,4 @@ export { useCancelTaskRun } from './useCancelTaskRun'; export { useCreateStandardTaskRun } from './useCreateStandardTaskRun'; +export { useRetryFailedTaskStart } from './useRetryFailedTaskStart'; export { useRouteHomeTask } from './useRouteHomeTask'; diff --git a/apps/web/src/hooks/task-runs/useRetryFailedTaskStart.ts b/apps/web/src/hooks/task-runs/useRetryFailedTaskStart.ts new file mode 100644 index 000000000..cfd52d7bb --- /dev/null +++ b/apps/web/src/hooks/task-runs/useRetryFailedTaskStart.ts @@ -0,0 +1,43 @@ +'use client'; + +import { useMutation, useQueryClient } from '@tanstack/react-query'; +import { toast } from 'sonner'; + +import { useTRPC } from '@/trpc/client'; + +export function useRetryFailedTaskStart(options?: { + onSuccess?: (result: { runId: number; taskId: string }) => void; + onError?: (error: unknown) => void; +}) { + const queryClient = useQueryClient(); + const trpc = useTRPC(); + + return useMutation( + trpc.taskRuns.retryFailedStart.mutationOptions({ + onSuccess: (result) => { + if (result.success) { + queryClient.invalidateQueries({ + queryKey: trpc.sandboxSession.byTaskId.queryKey(), + }); + + options?.onSuccess?.({ + runId: result.runId, + taskId: result.taskId, + }); + } else { + toast.error('Oops', { description: result.error }); + options?.onError?.(new Error(result.error)); + } + }, + onError: (error) => { + toast.error('Oops', { + description: + error instanceof Error + ? error.message + : 'An unknown error occurred', + }); + options?.onError?.(error); + }, + }), + ); +} diff --git a/apps/web/src/trpc/commands/sandbox-session/index.ts b/apps/web/src/trpc/commands/sandbox-session/index.ts index 3cf1a4986..776393f52 100644 --- a/apps/web/src/trpc/commands/sandbox-session/index.ts +++ b/apps/web/src/trpc/commands/sandbox-session/index.ts @@ -662,7 +662,21 @@ export async function getSandboxSessionByTaskIdCommand( const hasMessages = !!bootFailureMessage; const hasHarnessMessages = !!firstHarnessMessage; - const prompt = getTaskRunVisiblePrompt(taskRun); + const payloadPrompt = getTaskRunVisiblePrompt(taskRun); + // Fall back to durable task-level prompt when the run payload has no visible + // prompt, so failure UIs can still show the original request. + const prompt: { + text?: string; + images?: string[]; + visibleInTranscript: boolean; + } | null = + payloadPrompt ?? + (typeof task.prompt === 'string' && task.prompt.trim().length > 0 + ? { + text: task.prompt, + visibleInTranscript: true, + } + : null); const hasInitialPrompt = Boolean(prompt?.text?.trim()) || Boolean(prompt?.images?.length); diff --git a/apps/web/src/trpc/commands/task-runs/index.ts b/apps/web/src/trpc/commands/task-runs/index.ts index 58b13f652..2b0c1435b 100644 --- a/apps/web/src/trpc/commands/task-runs/index.ts +++ b/apps/web/src/trpc/commands/task-runs/index.ts @@ -485,3 +485,5 @@ export async function cancelTaskRunCommand( }; } } + +export { retryFailedTaskStartCommand } from './retry-failed-start'; diff --git a/apps/web/src/trpc/commands/task-runs/retry-failed-start.ts b/apps/web/src/trpc/commands/task-runs/retry-failed-start.ts new file mode 100644 index 000000000..317401ad9 --- /dev/null +++ b/apps/web/src/trpc/commands/task-runs/retry-failed-start.ts @@ -0,0 +1,102 @@ +import { + enqueueTaskRelaunch, + isRelaunchableFailedStartPayloadKind, +} from '@roomote/cloud-agents/server'; +import { and, db, desc, eq, isNull, taskRuns, tasks } from '@roomote/db/server'; +import { RunStatus, TaskPayloadKind } from '@roomote/types'; + +import type { UserAuthSuccess } from '@/types'; + +type RetryFailedTaskStartResult = + | { success: true; runId: number; taskId: string } + | { success: false; error: string }; + +/** + * Re-enqueues a failed first-start run on the same task so the user can retry + * after fixing provider capacity (for example Modal spend limits). + */ +export async function retryFailedTaskStartCommand( + auth: UserAuthSuccess, + input: { taskId: string; runId?: number }, +): Promise { + try { + const task = await db.query.tasks.findFirst({ + where: and(eq(tasks.id, input.taskId), isNull(tasks.deletedAt)), + columns: { id: true }, + }); + + if (task == null) { + return { success: false, error: 'Task not found' }; + } + + const failedRun = + input.runId !== undefined + ? await db.query.taskRuns.findFirst({ + where: and( + eq(taskRuns.id, input.runId), + eq(taskRuns.taskId, input.taskId), + ), + columns: { + id: true, + status: true, + payloadKind: true, + }, + }) + : await db.query.taskRuns.findFirst({ + where: and( + eq(taskRuns.taskId, input.taskId), + eq(taskRuns.status, RunStatus.Failed), + ), + orderBy: [desc(taskRuns.id)], + columns: { + id: true, + status: true, + payloadKind: true, + }, + }); + + if (!failedRun) { + return { success: false, error: 'Failed task run not found' }; + } + + if (failedRun.status !== RunStatus.Failed) { + return { + success: false, + error: 'Only failed task starts can be retried.', + }; + } + + if (failedRun.payloadKind === TaskPayloadKind.SnapshotResume) { + return { + success: false, + error: 'Use Retry resume for snapshot resume failures.', + }; + } + + if (!isRelaunchableFailedStartPayloadKind(failedRun.payloadKind)) { + return { + success: false, + error: 'This task type does not support start retry yet.', + }; + } + + const relaunchedRun = await enqueueTaskRelaunch({ + sourceRunId: failedRun.id, + actingUserId: auth.userId, + }); + + return { + success: true, + runId: relaunchedRun.id, + taskId: relaunchedRun.taskId, + }; + } catch (error) { + console.error('retryFailedTaskStart error:', error); + + return { + success: false, + error: + error instanceof Error ? error.message : 'An unknown error occurred.', + }; + } +} diff --git a/apps/web/src/trpc/routers/_app.ts b/apps/web/src/trpc/routers/_app.ts index 958f1d754..a48fde9fa 100644 --- a/apps/web/src/trpc/routers/_app.ts +++ b/apps/web/src/trpc/routers/_app.ts @@ -91,6 +91,7 @@ import { routeHomeTaskCommand, createStandardTaskRunCommand, cancelTaskRunCommand, + retryFailedTaskStartCommand, } from '../commands/task-runs'; import { exchangeSlackOAuthCodeCommand, @@ -699,6 +700,17 @@ export const appRouter = createRouter({ .mutation(({ ctx: { auth }, input }) => cancelTaskRunCommand(auth, input), ), + + retryFailedStart: protectedProcedure + .input( + z.object({ + taskId: z.string(), + runId: z.number().int().optional(), + }), + ) + .mutation(({ ctx: { auth }, input }) => + retryFailedTaskStartCommand(auth, input), + ), }), github: createRouter({ diff --git a/apps/web/src/types/task.ts b/apps/web/src/types/task.ts index 535513915..1effcc105 100644 --- a/apps/web/src/types/task.ts +++ b/apps/web/src/types/task.ts @@ -44,6 +44,7 @@ export const taskSchema = z.object({ defaultBranch: z.string().nullable().optional(), // Conversation cargo (present when the full tasks row is returned, e.g. by // the by-id command; absent from trimmed list rows). + prompt: z.string().nullable().optional(), draftPrompt: z.string().nullable().optional(), slackChannelId: z.string().nullable().optional(), slackThreadTs: z.string().nullable().optional(), diff --git a/packages/cloud-agents/src/server/__tests__/enqueue-task.test.ts b/packages/cloud-agents/src/server/__tests__/enqueue-task.test.ts index d2203c242..00eae1e9e 100644 --- a/packages/cloud-agents/src/server/__tests__/enqueue-task.test.ts +++ b/packages/cloud-agents/src/server/__tests__/enqueue-task.test.ts @@ -19,6 +19,9 @@ import { type SnapshotResumeTask, RunStatus, TaskPayloadKind, + TASK_KICKOFF_MESSAGE_SOURCE, + ACP_ENVELOPE_EVENT_TYPES, + ROOMOTE_RUNTIME_TASK_MESSAGE_PROTOCOL, } from '@roomote/types'; import { db, @@ -26,6 +29,7 @@ import { inArray, tasks, taskRuns, + taskMessages, taskPullRequests, taskRunEvents, users, @@ -40,6 +44,7 @@ import { import { TaskRunQueue, enqueueTask, + enqueueTaskRelaunch, resolveQueueScope, type FreshTaskLaunch, } from '../task-run-queue'; @@ -393,6 +398,126 @@ describe('enqueueTask snapshot resume', () => { }); }); +describe('enqueueTaskRelaunch failed start', () => { + it('creates a new fresh run on the same task after a failed start', async () => { + const userId = await createUser(); + + const failedRun = await launchFresh({ + initiator: { kind: 'user', userId }, + workflow: 'standard', + surface: 'web', + trigger: 'manual', + }); + + await db + .update(taskRuns) + .set({ + status: RunStatus.Failed, + error: 'Workspace has exceeded its spend limit', + completedAt: new Date(), + }) + .where(eq(taskRuns.id, failedRun.id)); + + await db + .update(tasks) + .set({ state: 'failed' }) + .where(eq(tasks.id, failedRun.taskId)); + + const relaunchRun = await enqueueTaskRelaunch( + { + sourceRunId: failedRun.id, + actingUserId: userId, + }, + { enqueue: false }, + ); + + expect(relaunchRun.taskId).toBe(failedRun.taskId); + expect(relaunchRun.id).not.toBe(failedRun.id); + expect(relaunchRun.kind).toBe('fresh'); + expect(relaunchRun.sourceRunId).toBe(failedRun.id); + expect(relaunchRun.payloadKind).toBe(TaskPayloadKind.StandardTask); + expect(relaunchRun.status).toBe(RunStatus.Pending); + expect(relaunchRun.actingUserId).toBe(userId); + + const taskAfter = await db.query.tasks.findFirst({ + where: eq(tasks.id, failedRun.taskId), + }); + expect(taskAfter!.state).toBe('active'); + + const runsForTask = await db.query.taskRuns.findMany({ + where: eq(taskRuns.taskId, failedRun.taskId), + }); + expect(runsForTask).toHaveLength(2); + }); + + it('rejects relaunch when the source run is not failed', async () => { + const userId = await createUser(); + + const pendingRun = await launchFresh({ + initiator: { kind: 'user', userId }, + workflow: 'standard', + surface: 'web', + trigger: 'manual', + }); + + await expect( + enqueueTaskRelaunch( + { + sourceRunId: pendingRun.id, + actingUserId: userId, + }, + { enqueue: false }, + ), + ).rejects.toThrow('Only failed task starts can be retried'); + }); + + it('allows relaunch when only a provider kickoff message exists', async () => { + const userId = await createUser(); + + const failedRun = await launchFresh({ + initiator: { kind: 'user', userId }, + workflow: 'standard', + surface: 'slack', + trigger: 'message', + }); + + await db + .update(taskRuns) + .set({ + status: RunStatus.Failed, + error: 'Workspace has exceeded its spend limit', + completedAt: new Date(), + }) + .where(eq(taskRuns.id, failedRun.id)); + + await db.insert(taskMessages).values({ + runId: failedRun.id, + taskId: failedRun.taskId, + ts: Date.now(), + eventType: ACP_ENVELOPE_EVENT_TYPES.AssistantMessage, + role: 'assistant', + protocol: ROOMOTE_RUNTIME_TASK_MESSAGE_PROTOCOL, + contentBlocks: [{ type: 'text', text: 'Looking into it in Full Stack.' }], + metadata: { source: TASK_KICKOFF_MESSAGE_SOURCE }, + payload: { + text: 'Looking into it in Full Stack.', + source: TASK_KICKOFF_MESSAGE_SOURCE, + }, + }); + + const relaunchRun = await enqueueTaskRelaunch( + { + sourceRunId: failedRun.id, + actingUserId: userId, + }, + { enqueue: false }, + ); + + expect(relaunchRun.taskId).toBe(failedRun.taskId); + expect(relaunchRun.sourceRunId).toBe(failedRun.id); + }); +}); + describe('enqueueTask PR linkage', () => { it('inserts the task_pull_requests row inside the create transaction', async () => { const prTask = { diff --git a/packages/cloud-agents/src/server/task-run-queue.ts b/packages/cloud-agents/src/server/task-run-queue.ts index b6c27eea3..401c625fb 100644 --- a/packages/cloud-agents/src/server/task-run-queue.ts +++ b/packages/cloud-agents/src/server/task-run-queue.ts @@ -18,6 +18,8 @@ import { type TaskPhase, RunStatus, TaskPayloadKind, + TASK_KICKOFF_MESSAGE_SOURCE, + activeRunStatuses, DEFAULT_DELEGATED_KEEPALIVE_MS, DEFAULT_KEEPALIVE_MS, DEFAULT_LAUNCH_CODING_HARNESS, @@ -44,6 +46,7 @@ import { syncTaskStateFromRuns, taskPullRequests, taskRuns, + taskMessages, tasks, users, environments, @@ -1514,6 +1517,246 @@ async function enqueueFreshLaunch( return taskRun; } +/** + * Payload kinds that can be re-enqueued on the same task after a terminal start + * failure (for example Modal spend limit during sandbox create). + * Snapshot resumes keep their dedicated retry path. + */ +const RELAUNCHABLE_FAILED_START_PAYLOAD_KINDS: ReadonlySet = + new Set([ + TaskPayloadKind.StandardTask, + TaskPayloadKind.Scan, + TaskPayloadKind.SlackAppMention, + TaskPayloadKind.LinearAgentSession, + TaskPayloadKind.GithubPrReviewFollowUp, + TaskPayloadKind.McpRecommendations, + ]); + +export function isRelaunchableFailedStartPayloadKind( + payloadKind: TaskPayloadKind, +): boolean { + return RELAUNCHABLE_FAILED_START_PAYLOAD_KINDS.has(payloadKind); +} + +function reconstructFreshTaskFromFailedRun(sourceRun: TaskRun): FreshTask { + return { + type: sourceRun.payloadKind, + harness: sourceRun.harness ?? undefined, + computeProvider: sourceRun.vendor ?? undefined, + payload: { ...(sourceRun.payload as object) }, + } as FreshTask; +} + +/** + * Re-enqueues a failed first-start run on the same task (new run row, same task + * id). Used when environment creation fails before the session can start and the + * user retries after fixing provider capacity or configuration. + */ +export async function enqueueTaskRelaunch( + input: { + sourceRunId: number; + actingUserId: string | null; + }, + options: EnqueueTaskOptions = {}, +): Promise { + await assertDeploymentIsActive(); + await assertUserIsNotDeleted(input.actingUserId); + + const sourceRun = await db.query.taskRuns.findFirst({ + where: eq(taskRuns.id, input.sourceRunId), + }); + + if (!sourceRun) { + throw new Error( + `Source run ${input.sourceRunId} was not found for task relaunch.`, + ); + } + + if (sourceRun.status !== RunStatus.Failed) { + throw new Error('Only failed task starts can be retried.'); + } + + if (!isRelaunchableFailedStartPayloadKind(sourceRun.payloadKind)) { + throw new Error( + `Task type '${sourceRun.payloadKind}' does not support start retry.`, + ); + } + + if (!sourceRun.payload?.repo && !sourceRun.payload?.environmentId) { + throw new Error('Failed run has no workspace information to relaunch.'); + } + + const existingTask = await db.query.tasks.findFirst({ + where: and(eq(tasks.id, sourceRun.taskId), isNull(tasks.deletedAt)), + columns: { + id: true, + workflow: true, + harness: true, + }, + }); + + if (!existingTask) { + throw new Error( + `Task ${sourceRun.taskId} was not found for task relaunch.`, + ); + } + + // Provider kickoff/start rows (for example Slack “started”) are written + // before provisioning and must not block restart after a failed start. + const priorHarnessMessage = await db.query.taskMessages.findFirst({ + where: and( + eq(taskMessages.runId, sourceRun.id), + sql`coalesce(${taskMessages.metadata}->>'source', '') <> ${TASK_KICKOFF_MESSAGE_SOURCE}`, + ), + columns: { id: true }, + }); + + if (priorHarnessMessage) { + throw new Error( + 'Only failed environment starts can be restarted. This run already has task messages.', + ); + } + + const task = reconstructFreshTaskFromFailedRun({ + ...sourceRun, + harness: sourceRun.harness ?? existingTask.harness, + }); + + const workspace = resolveTaskWorkspace(task.payload); + + if ( + !('sourceControlProvider' in task.payload) || + !task.payload.sourceControlProvider + ) { + const resolvedProvider = await resolveWorkspaceSourceControlProvider( + db, + workspace, + ); + + if (resolvedProvider) { + task.payload.sourceControlProvider = resolvedProvider; + } + } + + const { initialPaths } = await resolveEnvironmentContext(task); + const resolvedHarness = await resolveRequestedHarness(task); + const targetHarness = resolvedHarness.harness; + const { task: taskWithHarnessOverrides } = resolveEffectiveHarnessModelState({ + task, + targetHarness, + isSnapshotResume: false, + sourceRunHarnessModelOverrides: undefined, + deploymentMetadata: resolvedHarness.deploymentMetadata, + deploymentTaskModelSettings: resolvedHarness.deploymentTaskModelSettings, + deploymentCodeReviewModelId: + resolvedHarness.deploymentCodeReviewModelId ?? null, + }); + + const resolvedTaskPolicy = resolveTaskRuntimePolicy({ + taskType: taskWithHarnessOverrides.type, + launchClass: options.launchClass, + appEnv: Env.APP_ENV ?? 'development', + defaultKeepaliveMs: DEFAULT_KEEPALIVE_MS, + delegatedKeepaliveMs: DEFAULT_DELEGATED_KEEPALIVE_MS, + sandboxTimeoutMs: TASK_TIMEOUT_MS, + }); + + const targetComputeProvider = resolveComputeProviderTarget( + task.computeProvider, + await resolveDefaultComputeProvider(), + ); + + const taskRun = await db.transaction(async (tx) => { + // Serialize concurrent retries on the same task so two retries cannot both + // observe an empty active-run set and insert separate pending runs. + await tx.execute( + sql`SELECT id FROM tasks WHERE id = ${existingTask.id} FOR UPDATE`, + ); + + const activeRun = await tx.query.taskRuns.findFirst({ + where: and( + eq(taskRuns.taskId, existingTask.id), + inArray(taskRuns.status, [...activeRunStatuses]), + ), + columns: { id: true }, + }); + + if (activeRun) { + throw new Error('This task already has an active run.'); + } + + const stillFailed = await tx.query.taskRuns.findFirst({ + where: and( + eq(taskRuns.id, sourceRun.id), + eq(taskRuns.status, RunStatus.Failed), + ), + columns: { id: true }, + }); + + if (!stillFailed) { + throw new Error('Only failed task starts can be retried.'); + } + + const [insertedRun] = await tx + .insert(taskRuns) + .values({ + taskId: existingTask.id, + kind: 'fresh', + sourceRunId: sourceRun.id, + payloadKind: taskWithHarnessOverrides.type, + actingUserId: options.skipInitialActingUser ? null : input.actingUserId, + status: options.initialStatus ?? RunStatus.Pending, + taskPhase: options.initialTaskPhase ?? null, + error: options.initialError ?? null, + ...(options.initialStatus === RunStatus.Dequeued + ? { dequeuedAt: new Date() } + : {}), + harness: targetHarness, + vendor: targetComputeProvider, + port: task.payload.port, + initialPaths, + payload: taskWithHarnessOverrides.payload, + keepaliveMs: resolvedTaskPolicy.keepaliveMs, + }) + .returning(); + + if (!insertedRun) { + throw new Error('Failed to create `task_runs` record.'); + } + + if (options.afterCreateInTransaction) { + await options.afterCreateInTransaction(tx, insertedRun); + } + + await syncTaskStateFromRuns(tx, existingTask.id); + + return insertedRun; + }); + + void captureEvent('task_created', { + ...(input.actingUserId ? { userId: input.actingUserId } : {}), + properties: { + taskType: taskRun.payloadKind, + workflow: existingTask.workflow, + harness: taskRun.harness ?? null, + computeProvider: taskRun.vendor ?? null, + relaunch: true, + sourceRunId: sourceRun.id, + }, + }); + + await pushRunOntoQueue({ + taskRun, + scope: resolveQueueScope({ + workflow: existingTask.workflow as TaskWorkflow, + payloadKind: taskRun.payloadKind, + }), + options, + }); + + return taskRun; +} + async function enqueueSnapshotResume( input: ResumeTaskLaunch, options: EnqueueTaskOptions,